Maintenance. Work toward removing transposed dual subscriber bug.

Author: neocoretechs

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

@@ -1,5 +1,7 @@
 package org.ros.internal.node.server;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.ros.address.AdvertiseAddress;
 import org.ros.address.BindAddress;
 import org.ros.internal.node.client.MasterClient;
@@ -28,10 +30,17 @@
 import java.util.concurrent.ScheduledExecutorService;
 
 /**
+ * SlaveServer is the remote execution endpoint.<br/> This class is reflected for its invokable
+ * methods and that is made available to remote clients as the callable remote procedures.<br/>
+ * A remote client can request connections, get collections of publishers and subscriber and information
+ * about the state of the bus including connections between publishers and subscribers. TcpRosServer is
+ * the class that does most of the work and is wrapped by this class, which is here to provide the subset of
+ * remotely invokable methods.
  * @author jg
  */
 public class SlaveServer extends RpcServer {
-
+  private static boolean DEBUG = true;
+  private static final Log log = LogFactory.getLog(SlaveServer.class);
   private final GraphName nodeName;
   private final MasterClient masterClient;
   private final TopicParticipantManager topicParticipantManager;
@@ -53,10 +62,14 @@ public SlaveServer(GraphName nodeName, BindAddress tcpRosBindAddress,
 		invokableMethods = new ServerInvokeMethod(this.getClass().getName(), 0);
 	} catch (ClassNotFoundException e) {
 		throw new IOException(e);
-	}
-   
+	}  
     this.tcpRosServer =
         new TcpRosServer(tcpRosBindAddress, tcpRosAdvertiseAddress, topicParticipantManager, serviceManager, executorService);
+    if(DEBUG) {
+    	log.info("ADDRESSES SlaveServer ctor:"+nodeName+" TCPBind:"+tcpRosBindAddress+", TCPRosAdv:"+tcpRosAdvertiseAddress+", RPCBind:"+rpcBindAddress+", RPCAdv:"+rpcAdvertiseAddress);
+    	log.info("MANAGERS SlaveServer ctor:"+nodeName+" MasterClient:"+master+" TopicParticipantManager:"+topicParticipantManager+
+    			" ServiceManager:"+serviceManager+" ParameterManager:"+parameterManager+" SchedulaedExecutorService:"+executorService);
+    }
   }
 
   public AdvertiseAddress getTcpRosAdvertiseAddress() {
@@ -89,7 +102,17 @@ public void shutdown() throws IOException {
   public List<Object> getBusStats(String callerId) {
     throw new UnsupportedOperationException();
   }
-
+  /**
+   * We are returning a list of ArrayList of strings. Each ArrayList of string will
+   * contain 5 elements representing subscriber or publisher bus information.<br/>
+   * The first element of each list is a monotonically increasing integer.<br/>
+   * The second element is the node identifier of the opposite party, sub for pub and pub for sub.<br/>
+   * The third element is o for subscriber, i for publisher<br/>
+   * The fourth element is the protocol<br/>
+   * The fifth element is the topic name.<br/>
+   * @param callerId
+   * @return
+   */
   public List<Object> getBusInfo(String callerId) {
     List<Object> busInfo = new ArrayList<Object>();
     // The connection ID field is opaque to the user. A monotonically increasing
@@ -159,7 +182,12 @@ public Collection<DefaultPublisher<?>> getPublications() {
   public int paramUpdate(GraphName parameterName, Object parameterValue) {
     return parameterManager.updateParameter(parameterName, parameterValue);
   }
-
+  /**
+   * 
+   * @param callerId
+   * @param topicName
+   * @param publisherUris
+   */
   public void publisherUpdate(String callerId, String topicName, Collection<InetSocketAddress> publisherUris) {
     GraphName graphName = GraphName.of(topicName);
     if (topicParticipantManager.hasSubscriber(graphName)) {
@@ -168,6 +196,11 @@ public void publisherUpdate(String callerId, String topicName, Collection<InetSo
       Collection<PublisherIdentifier> identifiers =
           PublisherIdentifier.newCollectionFromUris(publisherUris, topicDeclaration);
       subscriber.updatePublishers(identifiers);
+      if(DEBUG) {
+    	  log.info("Updating subscriber:"+subscriber);
+    	  for(InetSocketAddress i: publisherUris)
+    		  log.info("Publisher:"+i);
+      }
     }
   }
   /**
@@ -183,13 +216,17 @@ public ProtocolDescription requestTopic(String topicName, Collection<String> pro
     // TODO(damonkohler): Use NameResolver.
     // Canonicalize topic name.
     GraphName graphName = GraphName.of(topicName).toGlobal();
+    if( DEBUG )
+    	log.info("Requesting topic:"+topicName+" for GraphName:"+graphName);
     if (!topicParticipantManager.hasPublisher(graphName)) {
       //throw new ServerException("No publishers for topic: " + graphName);
     	return null;
     }
     for (String protocol : protocols) {
       if (protocol.equals(ProtocolNames.TCPROS)) {
         try {
+        	if( DEBUG )
+        	  log.info("Requested topic:"+topicName+" for GraphName:"+graphName+" returning:"+tcpRosServer.getAdvertiseAddress());
           return new TcpRosProtocolDescription(tcpRosServer.getAdvertiseAddress());
         } catch (Exception e) {
           throw new ServerException(e);

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

@@ -37,7 +37,7 @@
  */
 public class DefaultPublisher<T> extends DefaultTopicParticipant implements Publisher<T> {
 
-  private static final boolean DEBUG = false;
+  private static final boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(DefaultPublisher.class);
 
   /**

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

@@ -16,7 +16,7 @@
  */
 public class SubscriberHandshake extends BaseClientHandshake {
 
-  private static final boolean DEBUG = false;
+  private static final boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(SubscriberHandshake.class);
 
   public SubscriberHandshake(ConnectionHeader outgoingConnectionHeader) {

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 = false;
+  private static boolean DEBUG = true;
   private static final Log log = LogFactory.getLog(SubscriberHandshakeHandler.class);
 
   private final IncomingMessageQueue<T> incomingMessageQueue;
@@ -38,15 +38,15 @@ public SubscriberHandshakeHandler(ConnectionHeader outgoingConnectionHeader,
     super(new SubscriberHandshake(outgoingConnectionHeader), executorService);
     this.incomingMessageQueue = incomingMessageQueue;
     if( DEBUG )
-    	log.info("subscriberhandshakeHandler ctor:"+this);
+    	log.info("SubscriberHandshakeHandler ctor:"+this+" IncomingMessageQueue:"+incomingMessageQueue+" outgoingConnectionHeader:"+outgoingConnectionHeader);
   }
   /**
    * Triggered from BaseClientHandshakeHandler channelRead
    */
   @Override
   protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandlerContext ctx) {
 	if( DEBUG )
-		log.info("SubscriberHandshakeHandler.onSuccess:"+ctx+" "+incomingConnectionHeader);
+		log.info("SubscriberHandshakeHandler onSuccess for:"+this+" ChannalHandlerContext:"+ctx+" incomingConnectionHeader:"+incomingConnectionHeader);
     ChannelPipeline pipeline = ctx.pipeline();
     pipeline.remove(getName());
     NamedChannelHandler namedChannelHandler = incomingMessageQueue.getMessageReceiver();
@@ -62,7 +62,7 @@ protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandl
    */
   @Override
   protected void onFailure(String errorMessage, ChannelHandlerContext ctx) throws IOException {
-    log.info("Subscriber handshake failed: " + errorMessage);
+    log.info("Subscriber handshake failed for:"+this+" ChannelHandlerContext:"+ctx+" with error:" + errorMessage);
     ctx.setReady(false);
     ctx.close();
   }
@@ -73,32 +73,30 @@ public String getName() {
   }
 
 
-  public void exceptionCaught(ChannelHandlerContext arg0, Throwable arg1) throws Exception {
-	onFailure(arg1.getMessage(), arg0);
+  public void exceptionCaught(ChannelHandlerContext ctx, Throwable arg1) throws Exception {
+	onFailure(arg1.getMessage(), ctx);
 	if( DEBUG )
-		log.info("SubscriberHandshakeHandler.exception caught:"+arg0+" "+arg1);
+		log.info("SubscriberHandshakeHandler exceptionCaught for:"+this+" ChannelHandlerContext:"+ctx+" error:"+arg1);
   }
 
   @Override
-  public void handlerAdded(ChannelHandlerContext arg0) throws Exception {
+  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 	if( DEBUG )
-		log.info("SubscriberHandshakeHandler.handlerAdded:"+arg0);
+		log.info("SubscriberHandshakeHandler handlerAdded for:"+this+" ChannelHandlerContext:"+ctx);
   }
 
   @Override
-  public void handlerRemoved(ChannelHandlerContext arg0) throws Exception {
+  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
 	if( DEBUG )
-		log.info("SubscriberHandshakeHandler.handlerRemoved:"+arg0);
+		log.info("SubscriberHandshakeHandler handlerRemoved for:"+this+" ChannelHandlerContext:"+ctx);
 
   }
 
-@Override
-public void userEventTriggered(ChannelHandlerContext ctx, Object event)
-		throws Exception {
+  @Override
+  public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
 	if( DEBUG )
-		log.info("SubscriberHandshakeHandler.userEventTriggered:"+ctx+" "+event);
-	
-}
+		log.info("SubscriberHandshakeHandler userEventTriggered for:"+this+" ChannelHandlerContext"+ctx+" "+event);
+  }
 
 
 }

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

@@ -1,19 +1,3 @@
-/*
- * Copyright (C) 2011 Google Inc.
- * 
- * Licensed under the Apache License, Version 2.0 (the "License"); you may not
- * use this file except in compliance with the License. You may obtain a copy of
- * the License at
- * 
- * http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations under
- * the License.
- */
-
 package org.ros.internal.node.topic;
 
 import org.apache.commons.logging.Log;
@@ -34,9 +18,10 @@
  * 
  * @author kwc@willowgarage.com (Ken Conley)
  * @author damonkohler@google.com (Damon Kohler)
+ * @author jg
  */
 public class TopicParticipantManager {
-
+	private static boolean DEBUG = true;
 	private static Log log = LogFactory.getLog(TopicParticipantManager.class);
   /**
    * A mapping from topic name to {@link Subscriber}.
@@ -91,6 +76,8 @@ public DefaultSubscriber<?> getSubscriber(GraphName topicName) {
   }
 
   public void addPublisher(DefaultPublisher<?> publisher) {
+	  if(DEBUG)
+		  log.info("Adding publisher:"+publisher+" to "+this);
     publishers.put(publisher.getTopicName(), publisher);
     if (listener != null) {
       listener.onPublisherAdded(publisher);
@@ -105,6 +92,8 @@ public void removePublisher(DefaultPublisher<?> publisher) {
   }
 
   public void addSubscriber(DefaultSubscriber<?> subscriber) {
+	  if(DEBUG)
+		  log.info("Adding subscriber:"+subscriber+" to "+this);
     subscribers.put(subscriber.getTopicName(), subscriber);
     if (listener != null) {
       listener.onSubscriberAdded(subscriber);
@@ -119,7 +108,8 @@ public void removeSubscriber(DefaultSubscriber<?> subscriber) {
   }
 
   public void addSubscriberConnection(DefaultSubscriber<?> subscriber, PublisherIdentifier publisherIdentifier) {
-	  //log.info("sub:"+subscriber+" pub:"+publisherIdentifier);
+	  if(DEBUG)
+		  log.info("Connecting subscriber:"+subscriber+" to publisher Identifier:"+publisherIdentifier+" for "+this);
 	  List<PublisherIdentifier> pubs = subscriberConnections.get(subscriber);
 	  //for(PublisherIdentifier p: pubs) log.info("Pub: "+p);
 	  if( pubs == null ) {
@@ -140,6 +130,8 @@ public void removeSubscriberConnection(DefaultSubscriber<?> subscriber, Publishe
   }
 
   public void addPublisherConnection(DefaultPublisher<?> publisher, SubscriberIdentifier subscriberIdentifier) {
+	  if(DEBUG)
+		  log.info("Connecting publisher:"+publisher+" to subscriber Identifier:"+subscriberIdentifier+" for "+this);
 	  List<SubscriberIdentifier> subs = publisherConnections.get(publisher);
 	  if( subs == null) {
 		  subs = new ArrayList<SubscriberIdentifier>();

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

@@ -20,7 +20,7 @@
  * @author jg
  */
 class UpdatePublisherRunnable<MessageType> implements Runnable {
-
+  private static boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(UpdatePublisherRunnable.class);
 
   private final DefaultSubscriber<MessageType> subscriber;
@@ -46,10 +46,13 @@ public UpdatePublisherRunnable(DefaultSubscriber<MessageType> subscriber,
   public void run() {
     SlaveClient slaveClient;
     try {
-      log.info("Attempting to create SlaveClient:"+nodeIdentifier.getName()+" pub:"+publisherIdentifier.getNodeUri());
+    	if(DEBUG)
+    		log.info("Attempting to create SlaveClient:"+nodeIdentifier.getName()+" pub:"+publisherIdentifier.getNodeUri());
       slaveClient = new SlaveClient(nodeIdentifier.getName(), publisherIdentifier.getNodeUri());
-      log.info("Slave client created "+nodeIdentifier.getName()+" pub:"+publisherIdentifier.getNodeUri());
-      log.info("Requesting topic name "+subscriber.getTopicName());
+      	if(DEBUG) {
+      		log.info("SlaveClient created "+nodeIdentifier.getName()+" pub:"+publisherIdentifier.getNodeUri());
+      		log.info("Requesting topic name "+subscriber.getTopicName());
+      	}
       Response<ProtocolDescription> response =
           slaveClient.requestTopic(subscriber.getTopicName(), ProtocolNames.SUPPORTED);
       // If null there is no publisher for the requested topic

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

@@ -23,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;
@@ -46,13 +46,13 @@ public void addListener(ClientHandshakeListener clientHandshakeListener) {
   public void channelActive(final ChannelHandlerContext ctx) throws Exception {
 	//final ByteBuffer bb = MessageBuffers.dynamicBuffer();
 	if( DEBUG ) {
-		log.info("Preparing OutgoingConnectionHeader:"+clientHandshake.getOutgoingConnectionHeader());
+		log.info("BaseClientHandshakeHandler channelActive for ChannelHandlerContext:"+ctx+" for:"+this+" Preparing OutgoingConnectionHeader:"+clientHandshake.getOutgoingConnectionHeader());
 	}
 
 	ctx.write(clientHandshake.getOutgoingConnectionHeader());
 
 	if( DEBUG )
-		log.info("BaseClientHandshakeHandler channelActive "+ctx+" OutgoingConnectionHeader reply to master complete");	
+		log.info("BaseClientHandshakeHandler channelActive for ChannelHandlerContext "+ctx+" OutgoingConnectionHeader reply to master complete for:"+this);	
 	/*
 	Utility.serialize(clientHandshake.getOutgoingConnectionHeader(), bb);
 	
@@ -73,7 +73,7 @@ public void failed(Throwable arg0, Void arg1) {
 
   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-  	log.info("Channel inactive"+ctx);
+  	log.info("Channel inactive ChannelHandlerContext:"+ctx+" for:"+this);
   }
 
   @Override
@@ -92,7 +92,7 @@ public Object channelRead(ChannelHandlerContext ctx, Object buff) throws Excepti
   @Override
   public void channelReadComplete(ChannelHandlerContext arg0) throws Exception {
   	if( DEBUG )
-  		log.info("SubscriberHandshakeHandler.channelReadComplete:"+arg0);
+  		log.info("SubscriberHandshakeHandler.channelReadComplete:"+arg0+" for:"+this);
 
   }
   /**

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

@@ -39,7 +39,7 @@ public class ChannelHandlerContextImpl implements ChannelHandlerContext {
 	/*Asynchronous*/Socket/*Channel*/ channel;
 	ChannelPipeline pipeline;
 	boolean ready = false;
-	Object mutex = new Object();
+	private Object mutex = new Object();
 	Set<String> outboundMessageTypes;
 	InputStream is = null;
 	OutputStream os = null;
@@ -188,13 +188,15 @@ public boolean isReady() {
 	 * Object to synchronize read and write completion for the channel in this context, since we will have
 	 * multiple outbound writers accessing the same channel
 	 */
+	@Override
 	public Object getChannelCompletionMutex() { return mutex; }
 
 	/**
 	 * Get the type of messages we want to send to the attached subscriber, based on the handshakes
 	 * received.
 	 * @return The HashSet of message type as String
 	 */
+	@Override
 	public Set<String> getMessageTypes() { return outboundMessageTypes; }
 
 

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

@@ -20,15 +20,15 @@
  */
 public class ChannelPipelineImpl implements ChannelPipeline {
 
-	LinkedBlockingDeque<Entry<String, ChannelHandler>> queue = new LinkedBlockingDeque<Entry<String, ChannelHandler>>();
+	private LinkedBlockingDeque<Entry<String, ChannelHandler>> queue = new LinkedBlockingDeque<Entry<String, ChannelHandler>>();
 	private ChannelHandlerContext ctx;
 
 	public ChannelPipelineImpl(ChannelHandlerContext ctx) {
 		this.ctx = ctx;
 	}
 	//public ChannelPipelineImpl() {}
 
-	public void setContext(ChannelHandlerContext ctx) { this.ctx = ctx; }
+	//public void setContext(ChannelHandlerContext ctx) { this.ctx = ctx; }
 
 
 	@Override