Merge pull request apache#106 from vrozov/v3.1.0

APEX-29 #resolve Use DefaultEventLoop.createEventLoop factory
brennonyork · Aug 4, 2015 · a3e9dfa · a3e9dfa
2 parents 66a75e0 + 8ae64ab
commit a3e9dfa
Show file tree

Hide file tree

Showing 10 changed files with 23 additions and 18 deletions.
diff --git a/api/pom.xml b/api/pom.xml
@@ -74,7 +74,7 @@
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>netlet</artifactId>
-      <version>1.1.0</version>
+      <version>1.2.0-SNAPSHOT</version>
     </dependency>
   </dependencies>
 </project>
diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/server/Server.java
@@ -144,7 +144,7 @@ public static void main(String[] args) throws Exception
       port = 0;
     }
 
-    DefaultEventLoop eventloop = new DefaultEventLoop("alone");
+    DefaultEventLoop eventloop = DefaultEventLoop.createEventLoop("alone");
     eventloop.start(null, port, new Server(port));
     new Thread(eventloop).start();
   }

diff --git a/bufferserver/src/main/java/com/datatorrent/bufferserver/util/System.java b/bufferserver/src/main/java/com/datatorrent/bufferserver/util/System.java
@@ -36,7 +36,7 @@ public static void startup(String identifier)
       DefaultEventLoop el = eventloops.get(identifier);
       if (el == null) {
         try {
-          eventloops.put(identifier, el = new DefaultEventLoop(identifier));
+          eventloops.put(identifier, el = DefaultEventLoop.createEventLoop(identifier));
         }
         catch (IOException io) {
           throw new RuntimeException(io);

diff --git a/bufferserver/src/test/java/com/datatorrent/bufferserver/client/SubscriberTest.java b/bufferserver/src/test/java/com/datatorrent/bufferserver/client/SubscriberTest.java
@@ -52,8 +52,8 @@ public class SubscriberTest
   public static void setupServerAndClients() throws Exception
   {
     try {
-      eventloopServer = new DefaultEventLoop("server");
-      eventloopClient = new DefaultEventLoop("client");
+      eventloopServer = DefaultEventLoop.createEventLoop("server");
+      eventloopClient = DefaultEventLoop.createEventLoop("client");
     }
     catch (IOException ioe) {
       throw new RuntimeException(ioe);

diff --git a/bufferserver/src/test/java/com/datatorrent/bufferserver/server/ServerTest.java b/bufferserver/src/test/java/com/datatorrent/bufferserver/server/ServerTest.java
@@ -58,8 +58,8 @@ public class ServerTest
   public static void setupServerAndClients() throws Exception
   {
     try {
-      eventloopServer = new DefaultEventLoop("server");
-      eventloopClient = new DefaultEventLoop("client");
+      eventloopServer = DefaultEventLoop.createEventLoop("server");
+      eventloopClient = DefaultEventLoop.createEventLoop("client");
     }
     catch (IOException ioe) {
       throw new RuntimeException(ioe);

diff --git a/bufferserver/src/test/java/com/datatorrent/bufferserver/storage/DiskStorageTest.java b/bufferserver/src/test/java/com/datatorrent/bufferserver/storage/DiskStorageTest.java
@@ -16,12 +16,10 @@
 package com.datatorrent.bufferserver.storage;
 
 import java.net.InetSocketAddress;
-import static java.lang.Thread.sleep;
 
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
 
 import com.datatorrent.bufferserver.packet.BeginWindowTuple;
 import com.datatorrent.bufferserver.packet.EndWindowTuple;
@@ -32,6 +30,9 @@
 import com.datatorrent.bufferserver.support.Subscriber;
 import com.datatorrent.netlet.DefaultEventLoop;
 
+import static java.lang.Thread.sleep;
+import static org.testng.Assert.assertEquals;
+
 /**
  *
  */
@@ -49,10 +50,10 @@ public class DiskStorageTest
   @BeforeClass
   public static void setupServerAndClients() throws Exception
   {
-    eventloopServer = new DefaultEventLoop("server");
+    eventloopServer = DefaultEventLoop.createEventLoop("server");
     eventloopServer.start();
 
-    eventloopClient = new DefaultEventLoop("client");
+    eventloopClient = DefaultEventLoop.createEventLoop("client");
     eventloopClient.start();
 
     instance = new Server(0, 1024,8);

diff --git a/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java b/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java
@@ -125,7 +125,7 @@ public class StreamingContainer extends YarnContainerMain
 
   static {
     try {
-      eventloop = new DefaultEventLoop("ProcessWideEventLoop");
+      eventloop = DefaultEventLoop.createEventLoop("ProcessWideEventLoop");
     }
     catch (IOException io) {
       throw new RuntimeException(io);

diff --git a/engine/src/main/java/com/datatorrent/stram/stream/FastPublisher.java b/engine/src/main/java/com/datatorrent/stram/stream/FastPublisher.java
@@ -189,6 +189,7 @@ public void activate(StreamContext context)
       if (!write) {
         key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
         write = true;
+        key.selector().wakeup();
       }
     }
   }
@@ -484,6 +485,7 @@ public void put(Object tuple)
     if (!write) {
       key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
       write = true;
+      key.selector().wakeup();
     }
   }
 

diff --git a/engine/src/test/java/com/datatorrent/stram/stream/FastStreamTest.java b/engine/src/test/java/com/datatorrent/stram/stream/FastStreamTest.java
@@ -16,7 +16,7 @@
 package com.datatorrent.stram.stream;
 
 import java.io.IOException;
-import static java.lang.Thread.sleep;
+
 import java.net.InetSocketAddress;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -36,6 +36,8 @@
 import com.datatorrent.stram.tuple.EndWindowTuple;
 import com.datatorrent.stram.tuple.Tuple;
 
+import static java.lang.Thread.sleep;
+
 /**
  *
  */
@@ -52,7 +54,7 @@ public class FastStreamTest
 
   static {
     try {
-      eventloop = new DefaultEventLoop("StreamTestEventLoop");
+      eventloop = DefaultEventLoop.createEventLoop("StreamTestEventLoop");
     }
     catch (IOException ex) {
       throw new RuntimeException(ex);

diff --git a/engine/src/test/java/com/datatorrent/stram/stream/SocketStreamTest.java b/engine/src/test/java/com/datatorrent/stram/stream/SocketStreamTest.java
@@ -19,8 +19,6 @@
 import com.datatorrent.stram.engine.StreamContext;
 import com.datatorrent.stram.engine.SweepableReservoir;
 import com.datatorrent.stram.support.StramTestSupport;
-import com.datatorrent.stram.stream.BufferServerPublisher;
-import com.datatorrent.stram.stream.BufferServerSubscriber;
 import com.datatorrent.stram.tuple.EndWindowTuple;
 import com.datatorrent.stram.tuple.Tuple;
 import com.datatorrent.api.Sink;
@@ -29,13 +27,15 @@
 import com.datatorrent.netlet.DefaultEventLoop;
 import com.datatorrent.netlet.EventLoop;
 import java.io.IOException;
-import static java.lang.Thread.sleep;
+
 import java.net.InetSocketAddress;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.lang.Thread.sleep;
+
 /**
  *
  */
@@ -49,7 +49,7 @@ public class SocketStreamTest
 
   static {
     try {
-      eventloop = new DefaultEventLoop("StreamTestEventLoop");
+      eventloop = DefaultEventLoop.createEventLoop("StreamTestEventLoop");
     }
     catch (IOException ex) {
       throw new RuntimeException(ex);