Fixed double subscriber ChannelHandlerContext bug. Tightened up and dekrufted code.

neocoretechs · neocoretechs · commit 2d7ab16d98da · 2020-01-15T14:13:32.000-08:00
diff --git a/src/main/java/org/ros/concurrent/EventDispatcher.java b/src/main/java/org/ros/concurrent/EventDispatcher.java
@@ -24,7 +24,6 @@
  *          the listener type
  */
 public class EventDispatcher<T> extends CancellableLoop {
-
   private final T listener;
   private final CircularBlockingDeque<SignalRunnable<T>> events;
 
@@ -35,11 +34,14 @@ public EventDispatcher(T listener, int queueCapacity) {
 
   public void signal(final SignalRunnable<T> signalRunnable) {
     events.addLast(signalRunnable);
+    //System.out.println("Event length addLast="+events.length()+" ***"+Thread.currentThread().getName());
   }
 
   @Override
   public void loop() throws InterruptedException {
     SignalRunnable<T> signalRunnable = events.takeFirst();
+    //System.out.println("Event length loop preRun="+events.length()+" ***"+Thread.currentThread().getName());
     signalRunnable.run(listener);
+    //System.out.println("Event length loop postRun="+events.length()+" ***"+Thread.currentThread().getName());
   }
 }
diff --git a/src/main/java/org/ros/exception/RemoteNotFoundException.java b/src/main/java/org/ros/exception/RemoteNotFoundException.java
@@ -0,0 +1,28 @@
+package org.ros.exception;
+
+import org.ros.internal.node.response.StatusCode;
+
+/**
+ * Remote exception indication a remote resource, such as a publisher, was not found and we want
+ * to toss a checked exception to handle that.
+ * @author jg Copyright (C) NeoCoreTechs 2018
+ */
+public class RemoteNotFoundException extends RosException {
+	private static final long serialVersionUID = 2514173639723076472L;
+
+	public RemoteNotFoundException(final Throwable throwable) {
+		super(throwable);
+	}
+
+	public RemoteNotFoundException(final String message, final Throwable throwable) {
+		super(message, throwable);
+	}
+
+	public RemoteNotFoundException(final String message) {
+		super(message);
+	}
+
+	public RemoteNotFoundException(StatusCode statusCode, String message) {
+		super(statusCode.toString()+" "+message);
+	}
+}
diff --git a/src/main/java/org/ros/internal/node/topic/DefaultPublisher.java b/src/main/java/org/ros/internal/node/topic/DefaultPublisher.java
@@ -65,6 +65,8 @@ public DefaultPublisher(NodeIdentifier nodeIdentifier, TopicDeclaration topicDec
     this.messageFactory = messageFactory;
     this.subscribers = arrayBlockingQueue;
     outgoingMessageQueue = new OutgoingMessageQueue<T>(executorService, arrayBlockingQueue);
+    if(DEBUG)
+    	log.info("DefaultPublisher contructed with "+outgoingMessageQueue.getNumberOfChannels()+" channels.");
     listeners = new ListenerGroup<PublisherListener<T>>(executorService);
     listeners.add(new DefaultPublisherListener<T>() {
       @Override
@@ -154,7 +156,7 @@ public ConnectionHeader finishHandshake(ConnectionHeader incomingHeader) {
     if (DEBUG) {
       //log.info("Subscriber handshake header: " + incomingHeader);
       //log.info("Publisher handshake header: " + topicDefinitionHeader);
-    	log.info("%%%%%%%%%%%%%%%% Handshake Complete %%%%%%%%%%%%%%%");
+    	log.info("%%%%%%%%%%%%%%%% Finishing Handshake with "+outgoingMessageQueue.getNumberOfChannels()+" channels. %%%%%%%%%%%%%%%");
     }
     // TODO(damonkohler): Return errors to the subscriber over the wire.
     String incomingType = incomingHeader.getField(ConnectionHeaderFields.TYPE);
@@ -195,11 +197,14 @@ public ConnectionHeader finishHandshake(ConnectionHeader incomingHeader) {
    */
   public void addSubscriber(SubscriberIdentifier subscriberIdentifer, ChannelHandlerContext ctx) {
     if (DEBUG) {
-      log.info(String.format("Adding subscriber %s channel %s to publisher %s.",
+      log.info(String.format("Adding subscriber %s ChannelHandlerContext %s to publisher %s.",
           subscriberIdentifer, ctx, this));
     }
     //outgoingMessageQueue.addChannel(ctx);
     subscribers.add(ctx);
+    if (DEBUG) {
+        log.info("Current number of subscribers:"+subscribers.size());
+    }
     signalOnNewSubscriber(subscriberIdentifer);
   }
 
diff --git a/src/main/java/org/ros/internal/node/topic/SubscriberHandshake.java b/src/main/java/org/ros/internal/node/topic/SubscriberHandshake.java
@@ -10,7 +10,7 @@
  * Handshake logic from the subscriber side of a topic connection.
  * The publisher receives the request for fields associated with the topic.
  * The subscriber receives the return from the publisher with the available fields and the
- * MD5 checksum to verifiy the message and complete the handshake.
+ * MD5 checksum to verify the message and complete the handshake.
  * 
  * @author jg
  */
diff --git a/src/main/java/org/ros/internal/node/topic/SubscriberHandshakeHandler.java b/src/main/java/org/ros/internal/node/topic/SubscriberHandshakeHandler.java
@@ -28,7 +28,7 @@
  *          the {@link Subscriber} may only subscribe to messages of this type
  */
 class SubscriberHandshakeHandler<T> extends BaseClientHandshakeHandler {
-  private static boolean DEBUG = true;
+  private static boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(SubscriberHandshakeHandler.class);
 
   private final IncomingMessageQueue<T> incomingMessageQueue;
diff --git a/src/main/java/org/ros/internal/transport/BaseClientHandshakeHandler.java b/src/main/java/org/ros/internal/transport/BaseClientHandshakeHandler.java
@@ -23,7 +23,7 @@
  * 
  */
 public abstract class BaseClientHandshakeHandler extends AbstractNamedChannelHandler {
-  protected static boolean DEBUG = true;
+  protected static boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(BaseClientHandshakeHandler.class);
   private final ClientHandshake clientHandshake;
   private final ListenerGroup<ClientHandshakeListener> clientHandshakeListeners;
diff --git a/src/main/java/org/ros/internal/transport/ChannelHandlerContextImpl.java b/src/main/java/org/ros/internal/transport/ChannelHandlerContextImpl.java
@@ -36,7 +36,6 @@ public class ChannelHandlerContextImpl implements ChannelHandlerContext {
 	private static final boolean DEBUG = false;
 	private static final Log log = LogFactory.getLog(ChannelHandlerContextImpl.class);
 	/*Asynchronous*/ChannelGroup channelGroup;
-	Executor executor;
 	/*Asynchronous*/Socket/*Channel*/ channel;
 	ChannelPipeline pipeline;
 	boolean ready = false;
@@ -45,10 +44,9 @@ public class ChannelHandlerContextImpl implements ChannelHandlerContext {
 	InputStream is = null;
 	OutputStream os = null;
 
-	public ChannelHandlerContextImpl(/*Asynchronous*/ChannelGroup channelGroup2, /*Asynchronous*/Socket channel2, Executor exc) {
+	public ChannelHandlerContextImpl(/*Asynchronous*/ChannelGroup channelGroup2, /*Asynchronous*/Socket channel2) {
 		channelGroup = channelGroup2;
 		channel = channel2;
-		executor = exc;
 		pipeline = new ChannelPipelineImpl(this);
 		outboundMessageTypes = (Set<String>) new HashSet<String>();
 	}
@@ -59,7 +57,7 @@ public void setChannel(/*Asynchronous*/Socket/*Channel*/ sock) {
 		
 	@Override
 	public Executor executor() {
-		return executor;
+		return channelGroup.getExecutorService();
 	}
 
 	@Override
@@ -172,7 +170,7 @@ public Socket channel() {
 
 	@Override
 	public String toString() {
-		return new String("ChannelHandlerContext:"+channel+" "+channelGroup+" "+executor+" "+pipeline+" ready:"+ready);
+		return new String("ChannelHandlerContext:"+channel+" ChannelGroup:"+channelGroup+" ChannelPipeline:"+pipeline+" ready:"+ready);
 	}
 
 	@Override
diff --git a/src/main/java/org/ros/internal/transport/queue/MessageDispatcher.java b/src/main/java/org/ros/internal/transport/queue/MessageDispatcher.java
@@ -70,9 +70,11 @@ public void addListener(MessageListener<T> messageListener, int limit) {
    * @return the newly allocated {@link SignalRunnable}
    */
   private SignalRunnable<MessageListener<T>> newSignalRunnable(final T lazyMessage) {
+	  //System.out.println("newSignalrunnable:"+lazyMessage+" ***"+Thread.currentThread().getName());
     return new SignalRunnable<MessageListener<T>>() {
       @Override
       public void run(MessageListener<T> messageListener) {
+    	//System.out.println("Run Signalrunnable:"+lazyMessage+" ***"+Thread.currentThread().getName());
         messageListener.onNewMessage(lazyMessage);
       }
     };
diff --git a/src/main/java/org/ros/internal/transport/queue/MessageReceiver.java b/src/main/java/org/ros/internal/transport/queue/MessageReceiver.java
@@ -35,7 +35,7 @@ public String getName() {
   @Override
   public Object channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
     if (DEBUG) {
-      log.info(String.format("Received message:"+msg));
+      log.info(String.format("Received message:"+msg+" ***"+Thread.currentThread().getName()));
     }
     lazyMessages.addLast((T) msg);
     return msg;
diff --git a/src/main/java/org/ros/internal/transport/queue/OutgoingMessageQueue.java b/src/main/java/org/ros/internal/transport/queue/OutgoingMessageQueue.java
@@ -33,7 +33,7 @@ public class OutgoingMessageQueue<T> {
   private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(OutgoingMessageQueue.class);
 
-  private static final int DEQUE_CAPACITY = 256;
+  private static final int DEQUE_CAPACITY = 16;
 
   private final CircularBlockingDeque<T> deque;
   private final Writer writer;
@@ -60,6 +60,8 @@ public void loop() throws InterruptedException {
       //  log.info(String.format("Writing %d bytes.", buffer.position()));
       //}
       final Iterator<ChannelHandlerContext> it = channels.iterator();
+      if(DEBUG)
+    	  log.info("Messaging "+channels.size()+" channels.");
       while(it.hasNext()) {
     	  final ChannelHandlerContext ctx = it.next();
     	  //final CountDownLatch cdl = new CountDownLatch(1);
@@ -70,7 +72,7 @@ public void loop() throws InterruptedException {
     	  if( ctx.isReady() && sendMessage ) {
     		  try {
     			if( DEBUG )
-    				log.info("Outgoing queue writing:"+message+" to "+ctx);
+    				log.info("Outgoing queue size="+deque.length()+" writing:"+message+" to "+ctx+" from "+Thread.currentThread().getName());
 				ctx.write(message);
 			} catch (IOException e) {
 				log.info("Closing failed write context:"+ctx+" due to "+e);
diff --git a/src/main/java/org/ros/internal/transport/tcp/AsynchBaseServer.java b/src/main/java/org/ros/internal/transport/tcp/AsynchBaseServer.java
@@ -21,20 +21,18 @@ public final class AsynchBaseServer extends AsynchTCPServer {
 	public InetSocketAddress address = null;
 	private TcpRosServer tcpserver = null;
 	AsynchTCPWorker uworker = null;
-	
-	
+		
 	public AsynchBaseServer(TcpRosServer tcpserver) throws IOException {
 		super();
 		this.address = tcpserver.getAddress();
 		this.tcpserver = tcpserver;
-		this.exc = tcpserver.getExecutor();
 		this.channelGroup = tcpserver.getChannelGroup();
 	}
 
 	public void startServer() throws IOException {
 		if( address == null )
 			throw new IOException("Server address not defined, can not start Base Server");
-		startServer(channelGroup, exc, address);
+		startServer(channelGroup, address);
 	}
 
 	
@@ -51,8 +49,10 @@ public void run() {
 				channel.setSendBufferSize(4096000);
 				channel.setReceiveBufferSize(4096000);
 				//channel.setTcpNoDelay(true);
-				ChannelHandlerContext ctx = new ChannelHandlerContextImpl(channelGroup, channel/*.get()*/, exc);
-				tcpserver.getSubscribers().add(ctx);
+				ChannelHandlerContext ctx = new ChannelHandlerContextImpl(channelGroup, channel/*.get()*/);
+				//if(DEBUG)
+				//	log.info("Adding new ChannelHandlerContext to subscribers array, subscribers="+tcpserver.getSubscribers().size()+" "+ctx);
+				//tcpserver.getSubscribers().add(ctx);
 				// inject the handlers, start handshake
 			    // inject calls initChannel on each ChannelInitializer in the factoryStack
 				tcpserver.getFactoryStack().inject(ctx);
@@ -65,10 +65,10 @@ public void run() {
 				// After it gets it the thread terminates and a new handler is inserted to generate outbound traffic
                 uworker = new AsynchTCPWorker(ctx);
                 // and send it all to executor for running
-                exc.execute(uworker);
+                channelGroup.getExecutorService().execute(uworker);
 				ctx.pipeline().fireChannelActive();
                 if( DEBUG ) {
-                    	log.info("ROS Asynch transport server worker starting for context:"+ctx);
+                    	log.info("AsynchBaseServer worker starting for context:"+ctx);
                 }   
 			} catch(Exception e) {
                     log.error("Asynch Server socket accept exception "+e,e);
diff --git a/src/main/java/org/ros/internal/transport/tcp/AsynchTCPServer.java b/src/main/java/org/ros/internal/transport/tcp/AsynchTCPServer.java
@@ -26,30 +26,28 @@ public abstract class AsynchTCPServer implements Cloneable, Runnable {
 	ServerSocket server = null;
 	ChannelGroup channelGroup;
 	Socket data = null;
-	Executor exc;
-	boolean shouldStop = false;
-	//public synchronized void startServer(AsynchronousChannelGroup group, Executor exc, int port) throws IOException {
-	public synchronized void startServer(ChannelGroup group, Executor exc, int port) throws IOException {	
-		this.exc = exc;
+	volatile boolean shouldStop = false;
+
+	public synchronized void startServer(ChannelGroup group, int port) throws IOException {	
 		if( server == null ) {
 			if( DEBUG )
 				log.info("AsynchTCPServer attempt local bind port "+port);
 			channelGroup = group;
 			//server = AsynchronousServerSocketChannel.open(channelGroup);
 			server = new ServerSocket();//channelGroup);
 			server.bind(new InetSocketAddress(port));
-			exc.execute(this);
+			group.getExecutorService().execute(this);
 		}
 	}
-	//public synchronized void startServer(AsynchronousChannelGroup group, Executor exc, InetSocketAddress binder) throws IOException {
-	public synchronized void startServer(ChannelGroup group, Executor exc, InetSocketAddress binder) throws IOException {
+	//public synchronized void startServer(AsynchronousChannelGroup group, InetSocketAddress binder) throws IOException {
+	public synchronized void startServer(ChannelGroup group, InetSocketAddress binder) throws IOException {
 		if( server == null ) {
 			if( DEBUG )
 				log.info("AsynchTCPServer attempt bind "+binder);
 			channelGroup = group;
 			server = new ServerSocket();
 			server.bind(binder);
-			exc.execute(this);
+			group.getExecutorService().execute(this);
 		}
 	}
 	public synchronized void stopServer() throws IOException {
diff --git a/src/main/java/org/ros/internal/transport/tcp/AsynchTCPWorker.java b/src/main/java/org/ros/internal/transport/tcp/AsynchTCPWorker.java
@@ -23,7 +23,7 @@ public class AsynchTCPWorker implements Runnable {
     public AsynchTCPWorker(ChannelHandlerContext ctx) throws IOException {
     	this.ctx = ctx;
     	if( DEBUG )
-    		log.info("AsynchTCPWorker constructed with context:"+ctx);
+    		log.info("AsynchTCPWorker constructed with ChannelHandlerContext:"+ctx);
 	}
 	
 	/**
@@ -34,28 +34,18 @@ public void run() {
 			try {
 				while(shouldRun) {
 					Object reso = ctx.read();
-					if( DEBUG )
-							log.info("ROS AsynchTCPWorker COMPLETED READ for "+ctx+"  Object:"+reso);
-					try {
-							ctx.pipeline().fireChannelRead(reso);
-					} catch (Exception e) {
-						if( DEBUG) {
-								log.info("Exception out of fireChannelRead",e);
-								e.printStackTrace();
-						}
-						ctx.pipeline().fireExceptionCaught(e);
-					}
+					//if( DEBUG )
+					//		log.info("AsynchTCPWorker COMPLETED READ for ChannelHandlerContext "+ctx+"  Result:"+reso);
+					ctx.pipeline().fireChannelRead(reso);
 					ctx.pipeline().fireChannelReadComplete();
 				} // shouldRun		
 			} catch(Exception se) {
-					log.error("AsynchTCPWorker terminating due to ",se);
+					log.error("AsynchTCPWorker terminating, ChannelHandlerContext "+ctx+" read failure due to ",se);
 					try {
 						ctx.pipeline().fireExceptionCaught(se);
 					} catch (Exception e) {}
 			} finally {
 				try {
-					if( DEBUG )
-						log.info("<<<<<<<<<< Datasocket closing >>>>>>>>");
 					ctx.setReady(false);
 					ctx.close();
 				} catch (IOException e) {}
diff --git a/src/main/java/org/ros/internal/transport/tcp/ChannelGroup.java b/src/main/java/org/ros/internal/transport/tcp/ChannelGroup.java
@@ -1,5 +1,8 @@
 package org.ros.internal.transport.tcp;
 
+import java.util.concurrent.ExecutorService;
+
 public interface ChannelGroup {
 	public void shutdown();
+	public ExecutorService getExecutorService();
 }
diff --git a/src/main/java/org/ros/internal/transport/tcp/ChannelGroupImpl.java b/src/main/java/org/ros/internal/transport/tcp/ChannelGroupImpl.java
@@ -2,7 +2,8 @@
 
 import java.util.concurrent.ExecutorService;
 /**
- * Manipulate groups of channels al la AsynchronousChannelGroup
+ * Manipulate groups of channels al la AsynchronousChannelGroup.
+ * Primarily contains the ExecutorService by which threads are put to work.
  * @author jg
  *
  */
@@ -14,4 +15,18 @@ public ChannelGroupImpl(ExecutorService executorService) {
 	public void shutdown() { 
 		executorService.shutdown();
 	}
+	public ExecutorService getExecutorService() { return executorService; }
+	
+	@Override
+	public boolean equals(Object o) {
+		return (executorService == ((ChannelGroup)o).getExecutorService());
+	}
+	@Override
+	public int hashCode() {
+		return executorService.hashCode();
+	}
+	@Override
+	public String toString() {
+		return "ChannelGroup executor "+executorService;
+	}
 }
diff --git a/src/main/java/org/ros/internal/transport/tcp/ChannelInitializerFactoryStack.java b/src/main/java/org/ros/internal/transport/tcp/ChannelInitializerFactoryStack.java
@@ -23,9 +23,13 @@ public class ChannelInitializerFactoryStack {
 	private LinkedBlockingDeque<ChannelInitializer> queue = new LinkedBlockingDeque<ChannelInitializer>();
 	
 	public void addFirst(ChannelInitializer ch) {
+		if(DEBUG)
+			log.info("Adding First ChannelInitializer "+ch+" queue size="+queue.size());
 		queue.addFirst(ch);
 	}
 	public void addLast(ChannelInitializer ch) {
+		if(DEBUG)
+			log.info("Adding Last ChannelInitializer "+ch+" queue size="+queue.size());
 		queue.addLast(ch);
 	}
 	/**
@@ -35,9 +39,14 @@ public void addLast(ChannelInitializer ch) {
 	 * @throws Exception
 	 */
 	public void inject(ChannelHandlerContext ctx) throws Exception {
+		if(DEBUG)
+			log.info("Injecting ChannelHandlerContext "+ctx+" queue size="+queue.size());
 		Iterator<ChannelInitializer> it = queue.iterator();
 		while(it.hasNext()) {
-			it.next().initChannel(ctx);
+			ChannelInitializer ch = it.next();
+			if(DEBUG)
+				log.info("ChannelInitializer.initChannel "+ch);
+			ch.initChannel(ctx);
 		}
 	}
 	
diff --git a/src/main/java/org/ros/internal/transport/tcp/TcpClient.java b/src/main/java/org/ros/internal/transport/tcp/TcpClient.java
diff --git a/src/main/java/org/ros/internal/transport/tcp/TcpClientManager.java b/src/main/java/org/ros/internal/transport/tcp/TcpClientManager.java
diff --git a/src/main/java/org/ros/internal/transport/tcp/TcpClientPipelineFactory.java b/src/main/java/org/ros/internal/transport/tcp/TcpClientPipelineFactory.java
diff --git a/src/main/java/org/ros/internal/transport/tcp/TcpRosServer.java b/src/main/java/org/ros/internal/transport/tcp/TcpRosServer.java
diff --git a/src/main/java/org/ros/internal/transport/tcp/TcpServerHandshakeHandler.java b/src/main/java/org/ros/internal/transport/tcp/TcpServerHandshakeHandler.java
diff --git a/src/test/java/org/ros/internal/transport/MessageQueueIntegrationTest.java b/src/test/java/org/ros/internal/transport/MessageQueueIntegrationTest.java

Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,6 @@`
`24`	`24`	`* the listener type`
`25`	`25`	`*/`
`26`	`26`	`public class EventDispatcher<T> extends CancellableLoop {`
`27`		`-`
`28`	`27`	`private final T listener;`
`29`	`28`	`private final CircularBlockingDeque<SignalRunnable<T>> events;`
`30`	`29`
`@@ -35,11 +34,14 @@ public EventDispatcher(T listener, int queueCapacity) {`
`35`	`34`
`36`	`35`	`public void signal(final SignalRunnable<T> signalRunnable) {`
`37`	`36`	`events.addLast(signalRunnable);`
	`37`	`+ //System.out.println("Event length addLast="+events.length()+" ***"+Thread.currentThread().getName());`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`@Override`
`41`	`41`	`public void loop() throws InterruptedException {`
`42`	`42`	`SignalRunnable<T> signalRunnable = events.takeFirst();`
	`43`	`+ //System.out.println("Event length loop preRun="+events.length()+" ***"+Thread.currentThread().getName());`
`43`	`44`	`signalRunnable.run(listener);`
	`45`	`+ //System.out.println("Event length loop postRun="+events.length()+" ***"+Thread.currentThread().getName());`
`44`	`46`	`}`
`45`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@`
`10`	`10`	`* Handshake logic from the subscriber side of a topic connection.`
`11`	`11`	`* The publisher receives the request for fields associated with the topic.`
`12`	`12`	`* The subscriber receives the return from the publisher with the available fields and the`
`13`		`- * MD5 checksum to verifiy the message and complete the handshake.`
	`13`	`+ * MD5 checksum to verify the message and complete the handshake.`
`14`	`14`	`*`
`15`	`15`	`* @author jg`
`16`	`16`	`*/`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@`
`23`	`23`	`*`
`24`	`24`	`*/`
`25`	`25`	`public abstract class BaseClientHandshakeHandler extends AbstractNamedChannelHandler {`
`26`		`- protected static boolean DEBUG = true;`
	`26`	`+ protected static boolean DEBUG = false;`
`27`	`27`	`private static final Log log = LogFactory.getLog(BaseClientHandshakeHandler.class);`
`28`	`28`	`private final ClientHandshake clientHandshake;`
`29`	`29`	`private final ListenerGroup<ClientHandshakeListener> clientHandshakeListeners;`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ public String getName() {`
`35`	`35`	`@Override`
`36`	`36`	`public Object channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {`
`37`	`37`	`if (DEBUG) {`
`38`		`- log.info(String.format("Received message:"+msg));`
	`38`	`+ log.info(String.format("Received message:"+msg+" ***"+Thread.currentThread().getName()));`
`39`	`39`	`}`
`40`	`40`	`lazyMessages.addLast((T) msg);`
`41`	`41`	`return msg;`