Merge pull request #57 from fluent/fix_ci

Fix CI test failures

Author: komamitsu

.travis.yml

@@ -5,6 +5,8 @@ jdk:
  - oraclejdk7
  - oraclejdk8
 
+sudo: false
+
 branches:
   only:
     - master

src/test/java/org/fluentd/logger/TestFluentLogger.java

@@ -72,6 +72,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
 
         FixedThreadManager threadManager = new FixedThreadManager(1);
         threadManager.submit(fluentd);
+        fluentd.waitUntilReady();
 
         // start loggers
         FluentLogger logger = FluentLogger.getLogger("testtag", host, port);
@@ -139,6 +140,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         });
         FixedThreadManager threadManager = new FixedThreadManager(1);
         threadManager.submit(fluentd);
+        fluentd.waitUntilReady();
 
         // start loggers
         FluentLogger[] loggers = new FluentLogger[loggerCount];
@@ -229,6 +231,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         threadManager.submit(fluentd1);
+        fluentd1.waitUntilReady();
 
         // start a logger
         final FluentLogger logger = FluentLogger.getLogger("testtag", host, port);
@@ -306,9 +309,8 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         threadManager.submit(fluentd2);
+        fluentd2.waitUntilReady();
 
-        // the logger should send an event successfully
-        TimeUnit.MILLISECONDS.sleep(500);
         {
             Map<String, Object> data = new HashMap<String, Object>();
             data.put("k5", "v5");
@@ -400,7 +402,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
 
         FixedThreadManager threadManager = new FixedThreadManager(1);
         threadManager.submit(fluentd);
-        Thread.sleep(1000);
+        fluentd.waitUntilReady();
 
         final FluentLogger logger = FluentLogger.getLogger(null, host, port);
         ExecutorService executorService = Executors.newFixedThreadPool(N);
@@ -487,7 +489,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
         }
 
         threadManager.submit(fluentd);
-        Thread.sleep(1000);
+        fluentd.waitUntilReady();
 
         // close loggers and it should flush the buffer
         logger.close();

src/test/java/org/fluentd/logger/sender/TestRawSocketSender.java

@@ -51,6 +51,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         fluentd.start();
+        fluentd.waitUntilReady();
 
         // start senders
         Sender sender = new RawSocketSender("localhost", port);
@@ -113,6 +114,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         fluentd.start();
+        fluentd.waitUntilReady();
 
         // start senders
         Sender sender = new RawSocketSender("localhost", port);
@@ -164,6 +166,8 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         fluentds[0].start();
+        fluentds[0].waitUntilReady();
+
         ports[1] = MockFluentd.randomPort();
         elists[1] = new ArrayList<Event>();
         fluentds[1] = new MockFluentd(ports[1], new MockFluentd.MockProcess() {
@@ -182,6 +186,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
             }
         });
         fluentds[1].start();
+        fluentds[1].waitUntilReady();
 
         // start senders
         Sender[] senders = new Sender[2];
@@ -274,6 +279,7 @@ public void process(MessagePack msgpack, Socket socket) throws IOException {
 
         MockFluentd fluentd = new MockFluentd(port, mockProcess);
         fluentd.start();
+        fluentd.waitUntilReady();
 
         Sender sender = new RawSocketSender("localhost", port);
         assertFalse(sender.isConnected());

src/test/java/org/fluentd/logger/util/MockFluentd.java

@@ -148,6 +148,17 @@ public void run() {
         _logger.debug("Terminated MockFluentd port:" + serverSocket.getLocalPort());
     }
 
+    public void waitUntilReady()
+            throws InterruptedException
+    {
+        int tick = 200;
+        while (!started.get()) {
+            TimeUnit.MILLISECONDS.sleep(tick);
+        }
+        // Just in case
+        TimeUnit.MILLISECONDS.sleep(tick);
+    }
+
     public void close() throws IOException {
         finished.set(true);
         service.shutdown();