Using blocking stream IO instead of NIO channels.

Author: neocoretechs

src/main/java/org/ros/internal/node/DefaultNode.java

@@ -167,7 +167,7 @@ public void run() {
       }
     });
   }
-
+  
   private void start() {
     // The Registrar must be started first so that master registration is
     // possible during startup.

src/main/java/org/ros/internal/node/client/RemoteClient.java

@@ -11,18 +11,14 @@
 import java.net.SocketException;
 import java.util.concurrent.ArrayBlockingQueue;
 
-
-
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.ros.internal.node.response.VoidResultFactory;
 import org.ros.internal.node.server.RemoteRequestInterface;
 import org.ros.internal.node.server.ThreadPoolManager;
-import org.ros.internal.transport.tcp.TcpRosServer;
+
 
 /**
- * This class functions as client to the  remote node.
+ * This class functions as client to the remote node.
  * @author jg
  * Copyright (C) NeoCoreTechs 2014,2015
  */
@@ -40,21 +36,12 @@ public class RemoteClient implements Runnable {
 	private boolean shouldRun = true; // master service thread control
 	private Object waitHalt = new Object(); 
 
-	private VoidResultFactory handleNull = new VoidResultFactory();
-	
 	ArrayBlockingQueue<RemoteRequestInterface> requests = new ArrayBlockingQueue<RemoteRequestInterface>(1);
 	ArrayBlockingQueue<Object> responses = new ArrayBlockingQueue<Object>(1);
 
 
 	/**
-	 * Start a relatrix client. Contact the boot time portion of server and queue a CommandPacket to open the desired
-	 * database and get back the master and slave ports of the remote server. The main client thread then
-	 * contacts the server master port, and the remote slave port contacts the master of the client. A WorkerRequestProcessor
-	 * thread is created to handle the processing of payloads and a comm thread handles the bidirectional traffic to server
-	 * @param dbName The name of the remote DB in full qualified path form
-	 * @param remote The remote database name for cluster server
-	 * @param bootNode The name of the remote server host
-	 * @param bootPort Then name of the remote host port on which RelatrixServer is running
+	 * Start a remote client.
 	 */
 	public RemoteClient(String bootNode, int bootPort)  throws IOException {
 		if( TEST ) {

src/main/java/org/ros/internal/node/server/SlaveServer.java

@@ -24,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 
 /**
@@ -65,7 +66,7 @@ public AdvertiseAddress getTcpRosAdvertiseAddress() {
    * Return the ChannelHandlerContext array of subscribers
    * @return
    */
-  public List<ChannelHandlerContext> getSubscribers() {
+  public ArrayBlockingQueue<ChannelHandlerContext> getSubscribers() {
 	  return tcpRosServer.getSubscribers();
   }
   /**

src/main/java/org/ros/internal/node/service/DefaultServiceClient.java

@@ -131,12 +131,17 @@ public void shutdown() {
 
   @Override
   public void call(T request, ServiceResponseListener<S> listener) {
-    ByteBuffer buffer = messageBufferPool.acquire();
+    //ByteBuffer buffer = messageBufferPool.acquire();
     // serialize request to buffer
-    Utility.serialize(request, buffer);
+    //Utility.serialize(request, buffer);
     responseListeners.add(listener);
-    tcpClient.write(buffer);
-    messageBufferPool.release(buffer);
+    try {
+		tcpClient.getContext().write(request);
+	} catch (IOException e) {
+		
+		e.printStackTrace();
+	}
+    //messageBufferPool.release(buffer);
   }
 
   @Override

src/main/java/org/ros/internal/node/service/ServiceClientHandshakeHandler.java

@@ -66,7 +66,7 @@ public String getName() {
 @Override
 public void handlerAdded(ChannelHandlerContext arg0) throws Exception {
 	if( DEBUG )
-		log.debug("Handler added "+arg0);
+		log.debug(this+" Handler added "+arg0);
 
 }
 

src/main/java/org/ros/internal/node/topic/DefaultPublisher.java

@@ -21,6 +21,7 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.Channel;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
@@ -55,15 +56,15 @@ public class DefaultPublisher<T> extends DefaultTopicParticipant implements Publ
   private final NodeIdentifier nodeIdentifier;
   private final MessageFactory messageFactory;
   // List of channelHandlerContexts that represent subscribers connected through publisher socket
-  private final List<ChannelHandlerContext> subscribers;
+  private final ArrayBlockingQueue<ChannelHandlerContext> subscribers;
 
   public DefaultPublisher(NodeIdentifier nodeIdentifier, TopicDeclaration topicDeclaration,
-		  MessageFactory messageFactory, ScheduledExecutorService executorService, List<ChannelHandlerContext> subscribers) throws IOException {
+		  MessageFactory messageFactory, ScheduledExecutorService executorService, ArrayBlockingQueue<ChannelHandlerContext> arrayBlockingQueue) throws IOException {
     super(topicDeclaration);
     this.nodeIdentifier = nodeIdentifier;
     this.messageFactory = messageFactory;
-    this.subscribers = subscribers;
-    outgoingMessageQueue = new OutgoingMessageQueue<T>(executorService, subscribers);
+    this.subscribers = arrayBlockingQueue;
+    outgoingMessageQueue = new OutgoingMessageQueue<T>(executorService, arrayBlockingQueue);
     listeners = new ListenerGroup<PublisherListener<T>>(executorService);
     listeners.add(new DefaultPublisherListener<T>() {
       @Override
@@ -147,7 +148,8 @@ public void publish(T message) {
    * 
    * @return encoded connection header from subscriber
    */
-  public ByteBuffer finishHandshake(ConnectionHeader incomingHeader) {
+  //public ByteBuffer finishHandshake(ConnectionHeader incomingHeader) {
+public ConnectionHeader finishHandshake(ConnectionHeader incomingHeader) {
     ConnectionHeader topicDefinitionHeader = getTopicDeclarationHeader();
     if (DEBUG) {
       //log.info("Subscriber handshake header: " + incomingHeader);
@@ -171,9 +173,16 @@ public ByteBuffer finishHandshake(ConnectionHeader incomingHeader) {
     // TODO(damonkohler): Force latch mode to be consistent throughout the life
     // of the publisher.
     outgoingConnectionHeader.addField(ConnectionHeaderFields.LATCHING, getLatchMode() ? "1" : "0");
+    return outgoingConnectionHeader;
+    /*
     ByteBuffer buffer = MessageBuffers.dynamicBuffer();
+    buffer.position(4);
     Utility.serialize(outgoingConnectionHeader, buffer);
+    buffer.rewind();
+    buffer.putInt(buffer.limit()-4); // put the TLV size
+    buffer.position(buffer.limit());
     return buffer;
+    */
   }
 
   /**

src/main/java/org/ros/internal/node/topic/DefaultSubscriber.java

@@ -129,7 +129,6 @@ public void addPublisher(PublisherIdentifier publisherIdentifier, InetSocketAddr
         return;
       }
       try {
-    	
 		tcpClientManager.connect(toString(), address);
       } catch (IOException e) {
     	log.error("Failure attempting to add publisher "+toString()+" "+address);

src/main/java/org/ros/internal/node/topic/PublisherFactory.java

@@ -9,6 +9,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 
 /**
@@ -41,21 +42,21 @@ public PublisherFactory(NodeIdentifier nodeIdentifier,
    *          the message type associated with the {@link Publisher}
    * @param topicDeclaration
    *          {@link TopicDeclaration} that is being published
- * @param subscribers 
+ * @param arrayBlockingQueue 
    * @param messageSerializer
    *          the {@link MessageSerializer} used for published messages
    * @return a new or cached {@link Publisher} instance
    * @throws IOException 
    */
   @SuppressWarnings("unchecked")
-  public <T> Publisher<T> newOrExisting(TopicDeclaration topicDeclaration, List<ChannelHandlerContext> subscribers) throws IOException {
+  public <T> Publisher<T> newOrExisting(TopicDeclaration topicDeclaration, ArrayBlockingQueue<ChannelHandlerContext> arrayBlockingQueue) throws IOException {
     GraphName topicName = topicDeclaration.getName();
     synchronized (mutex) {
       if (topicParticipantManager.hasPublisher(topicName)) {
         return (DefaultPublisher<T>) topicParticipantManager.getPublisher(topicName);
       } else {
         DefaultPublisher<T> publisher =
-            new DefaultPublisher<T>(nodeIdentifier, topicDeclaration, messageFactory, executorService, subscribers);
+            new DefaultPublisher<T>(nodeIdentifier, topicDeclaration, messageFactory, executorService, arrayBlockingQueue);
         publisher.addListener(new DefaultPublisherListener<T>() {
           @Override
           public void onNewSubscriber(Publisher<T> publisher, SubscriberIdentifier subscriberIdentifier) {

src/main/java/org/ros/internal/node/topic/SubscriberHandshakeHandler.java

@@ -8,8 +8,6 @@
 import org.ros.internal.transport.ConnectionHeader;
 import org.ros.internal.transport.ConnectionHeaderFields;
 import org.ros.internal.transport.queue.IncomingMessageQueue;
-import org.ros.internal.transport.tcp.AsynchTCPWorker;
-import org.ros.internal.transport.tcp.AsynchTempTCPWorker;
 import org.ros.internal.transport.tcp.NamedChannelHandler;
 import org.ros.node.topic.Publisher;
 import org.ros.node.topic.Subscriber;
@@ -30,7 +28,7 @@
  *          the {@link Subscriber} may only subscribe to messages of this type
  */
 class SubscriberHandshakeHandler<T> extends BaseClientHandshakeHandler {
-  private static boolean DEBUG = false;
+  private static boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(SubscriberHandshakeHandler.class);
 
   private final IncomingMessageQueue<T> incomingMessageQueue;
@@ -57,17 +55,7 @@ protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandl
     if (latching != null && latching.equals("1")) {
       incomingMessageQueue.setLatchMode(true);
     }
-    // the termination of the AsynchTempTCPWorker after successful handshake will set context ready
-    // Launch the permanent worker
-    AsynchTCPWorker uworker = null;
-	try {
-		uworker = new AsynchTCPWorker(ctx);
-	} catch (IOException e) {
-		log.error("Cannot start worker for context:"+ctx);
-		e.printStackTrace();
-		return;
-	}
-    executor.execute(uworker); 
+    ctx.setReady(true);
   }
   /**
    * Triggered from BaseClientHandshakeHandler

src/main/java/org/ros/internal/transport/BaseClientHandshakeHandler.java

@@ -9,8 +9,11 @@
 import org.ros.internal.transport.tcp.AbstractNamedChannelHandler;
 
 import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
+import java.nio.channels.SocketChannel;
 import java.util.concurrent.ExecutorService;
 
 /**
@@ -20,7 +23,7 @@
  * 
  */
 public abstract class BaseClientHandshakeHandler extends AbstractNamedChannelHandler {
-  protected static boolean DEBUG = false;
+  protected static boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(BaseClientHandshakeHandler.class);
   private final ClientHandshake clientHandshake;
   private final ListenerGroup<ClientHandshakeListener> clientHandshakeListeners;
@@ -40,21 +43,32 @@ public void addListener(ClientHandshakeListener clientHandshakeListener) {
   }
 
   @Override
-  public void channelActive(ChannelHandlerContext ctx) throws Exception {
-	ByteBuffer bb = MessageBuffers.dynamicBuffer();
+  public void channelActive(final ChannelHandlerContext ctx) throws Exception {
+	//final ByteBuffer bb = MessageBuffers.dynamicBuffer();
+	if( DEBUG ) {
+		log.info("Preparing OutgoingConnectionHeader:"+clientHandshake.getOutgoingConnectionHeader());
+	}
+	
+	ctx.write(clientHandshake.getOutgoingConnectionHeader());
+	
+	if( DEBUG )
+		log.info("BaseClientHandshakeHandler channelActive "+ctx+" OutgoingConnectionHeader reply to master complete");	
+	/*
 	Utility.serialize(clientHandshake.getOutgoingConnectionHeader(), bb);
+	
     ctx.write(bb, new CompletionHandler<Integer, Void>() {
 		@Override
 		public void completed(Integer arg0, Void arg1) {
 			if( DEBUG )
-				log.info("BaseClientHandshakeHandler channelActive reply to master complete");	
+				log.info("BaseClientHandshakeHandler channelActive "+ctx+" reply to master complete with "+arg0+" buffer:"+bb);	
 		}
 		@Override
 		public void failed(Throwable arg0, Void arg1) {
-			log.info("BaseClientHandshakeHandler channelActive reply to master failed with:"+arg0);	
+			log.info("BaseClientHandshakeHandler channelActive "+ctx+" reply to master failed with:"+arg0);	
 		}
     	
     });
+    */
   }
 
   @Override