Fixes to improve telegraf integration. (#103)

Author: vjkoskela

pom.xml

@@ -71,8 +71,8 @@
 
   <properties>
     <!--Dependency versions-->
-    <akka.version>2.5.6</akka.version>
-    <akka.http.version>10.0.11</akka.http.version>
+    <akka.version>2.5.13</akka.version>
+    <akka.http.version>10.1.3</akka.http.version>
     <apache.httpclient.version>4.5.1</apache.httpclient.version>
     <apache.httpcore.version>4.4.3</apache.httpcore.version>
     <arpnetworking.commons.version>1.16.0</arpnetworking.commons.version>

src/main/java/com/arpnetworking/metrics/common/sources/BaseTcpSource.java

@@ -72,7 +72,8 @@ protected BaseTcpSource(final Builder<?, ?> builder) {
 
         @Override
         public void preStart() {
-            _tcpManager.tell(
+            final ActorRef tcpManager = Tcp.get(getContext().system()).manager();
+            tcpManager.tell(
                     TcpMessage.bind(
                             getSelf(),
                             new InetSocketAddress(_host, _port),
@@ -87,24 +88,24 @@ public Receive createReceive() {
                         getSender().tell(_isReady, getSelf());
                     })
                     .match(Tcp.Bound.class, tcpBound -> {
-                        _isReady = true;
-                        _tcpManager.tell(tcpBound, getSelf());
                         LOGGER.info()
                                 .setMessage("Tcp server binding complete")
                                 .addData("name", _sink.getName())
                                 .addData("address", tcpBound.localAddress().getAddress().getHostAddress())
                                 .addData("port", tcpBound.localAddress().getPort())
                                 .log();
+
+                        _isReady = true;
                     })
                     .match(Tcp.CommandFailed.class, failed -> {
-                        getContext().stop(getSelf());
                         LOGGER.warn()
                                 .setMessage("Tcp server bad command")
                                 .addData("name", _sink.getName())
                                 .log();
+
+                        getContext().stop(getSelf());
                     })
                     .match(Tcp.Connected.class, tcpConnected -> {
-                        _tcpManager.tell(tcpConnected, getSelf());
                         LOGGER.debug()
                                 .setMessage("Tcp connection established")
                                 .addData("name", _sink.getName())
@@ -141,16 +142,13 @@ protected BaseTcpListenerActor(final BaseTcpSource source) {
             _host = source._host;
             _port = source._port;
             _acceptQueue = source._acceptQueue;
-
-            _tcpManager = Tcp.get(getContext().system()).manager();
         }
 
         private boolean _isReady = false;
         private final BaseTcpSource _sink;
         private final String _host;
         private final int _port;
         private final int _acceptQueue;
-        private final ActorRef _tcpManager;
 
         private static final String IS_READY = "IsReady";
     }

src/main/java/com/arpnetworking/metrics/mad/parsers/TelegrafJsonToRecordParser.java

@@ -258,7 +258,7 @@ public ZonedDateTime create(final long timestamp) {
         NANOSECONDS {
             @Override
             public ZonedDateTime create(final long timestamp) {
-                return ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp * 1000000), ZoneOffset.UTC);
+                return ZonedDateTime.ofInstant(Instant.ofEpochMilli(timestamp / 1000000), ZoneOffset.UTC);
             }
         };
 

src/test/java/com/arpnetworking/metrics/common/sources/StatsdSourceTest.java

@@ -66,6 +66,7 @@ public void tearDown() {
     }
 
     @Test
+    @SuppressWarnings("deprecation")
     public void test() {
         final StatsdSource statsdSource = new StatsdSource.Builder()
                 .setActorSystem(_actorSystem)