Minimal debug, cleaned up handshake

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

@@ -27,7 +27,7 @@
 import org.ros.internal.node.topic.SubscriberFactory;
 import org.ros.internal.node.topic.TopicDeclaration;
 import org.ros.internal.node.topic.TopicParticipantManager;
-import org.ros.internal.transport.tcp.TcpRosServer;
+
 import org.ros.message.MessageFactory;
 import org.ros.message.Time;
 import org.ros.namespace.GraphName;
@@ -62,7 +62,7 @@
  * 
  */
 public class DefaultNode implements ConnectedNode {
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(DefaultNode.class);
   /**
    * The maximum delay before shutdown will begin even if all

src/main/java/org/ros/internal/node/client/ParameterClient.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.client;
 
 
@@ -24,35 +8,30 @@
 import org.ros.internal.node.response.IntegerResultFactory;
 import org.ros.internal.node.response.ObjectResultFactory;
 import org.ros.internal.node.response.Response;
-import org.ros.internal.node.response.ResultFactory;
-import org.ros.internal.node.response.StringListResultFactory;
+
 import org.ros.internal.node.response.StringResultFactory;
 import org.ros.internal.node.response.VoidResultFactory;
 import org.ros.internal.node.rpc.MasterRpcEndpointImpl;
 import org.ros.internal.node.rpc.ParameterServerRpcEndpoint;
 import org.ros.internal.node.server.NodeIdentifier;
 import org.ros.internal.node.server.ParameterServer;
-import org.ros.internal.transport.tcp.TcpRosServer;
+
 import org.ros.namespace.GraphName;
 
 import java.io.IOException;
-import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.net.URI;
 import java.net.UnknownHostException;
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
+
 
 /**
  * Provide access to the RPC API for a ROS {@link ParameterServer}.
+ * The ParameterServer typically starts one port higher than the master.
  * 
- * @author [email protected] (Ken Conley)
- * @author [email protected] (Damon Kohler)
  * @author jg
  */
 public class ParameterClient extends Client<ParameterServerRpcEndpoint> {
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(ParameterClient.class);  
   private final NodeIdentifier nodeIdentifier;
   private final String nodeName;

src/main/java/org/ros/internal/node/client/Registrar.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.client;
 
 import org.apache.commons.logging.Log;
@@ -34,7 +18,7 @@
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.net.URI;
+
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.Callable;
@@ -45,12 +29,11 @@
  * Manages topic, and service registrations of a {@link SlaveServer} with the
  * {@link MasterServer}.
  * 
- * @author [email protected] (Ken Conley)
- * @author [email protected] (Damon Kohler)
+ * @author jg
  */
 public class Registrar implements TopicParticipantManagerListener, ServiceManagerListener {
 
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(Registrar.class);
 
   private static final int SHUTDOWN_TIMEOUT = 5;

src/main/java/org/ros/internal/node/rpc/SlaveRpcEndpointImpl.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.rpc;
 
 
@@ -35,16 +19,18 @@
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
+
 import java.util.Set;
-import java.util.Vector;
+
 
 /**
- * @author [email protected] (Damon Kohler)
+ * Facility for contacting the remote master and issuing commands via the remote invokable methods.
+ * The transport is accomplished via the RemoteRequestinterface implementors.
+ * @author jg
  */
 public class SlaveRpcEndpointImpl implements SlaveRpcEndpoint {
 
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(SlaveRpcEndpointImpl.class);
 
   //private final SlaveServer slave;

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

@@ -13,7 +13,7 @@
  *
  */
 public final class BaseServer extends TCPServer {
-	private static boolean DEBUG = true;
+	private static boolean DEBUG = false;
     private static final Log log = LogFactory.getLog(BaseServer.class);
 	public int WORKBOOTPORT = 8090;
 	public InetAddress address = null;

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

@@ -26,7 +26,7 @@
  * @author Groff
  */
 public class ParameterServer extends RpcServer {
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(ParameterServer.class);
 
   private final Map<String, Object> tree;

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

@@ -22,7 +22,7 @@
  */
 public abstract class RpcServer {
 
-  private static final boolean DEBUG = true;
+  private static final boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(RpcServer.class);
 
   private final AdvertiseAddress advertiseAddress;

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

@@ -9,7 +9,7 @@
 * TCPServer is the superclass of all objects using ServerSockets.
 */
 public abstract class TCPServer implements Cloneable, Runnable {
-	private static boolean DEBUG = true;
+	private static boolean DEBUG = false;
 	private static final Log log = LogFactory.getLog(TCPServer.class);
 	ServerSocket server = null;
 	Socket data = null;

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

@@ -17,7 +17,7 @@
  *
  */
 public class TCPWorker implements Runnable {
-	private static final boolean DEBUG = true;
+	private static final boolean DEBUG = false;
 	private static final Log log = LogFactory.getLog(TCPWorker.class);
 	public boolean shouldRun = true;
 	private Socket dataSocket;

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

@@ -26,7 +26,7 @@
  *
  */
 public class ThreadPoolManager {
-	private static final boolean DEBUG = true;
+	private static final boolean DEBUG = false;
 	  private static final Log log = LogFactory.getLog(ThreadPoolManager.class);
 	private static String DEFAULT_THREAD_POOL = "RPCSERVER";
 	private int threadNum = 0;

src/main/java/org/ros/internal/node/server/master/MasterRegistrationManagerImpl.java

@@ -26,7 +26,7 @@
  * @author jg
  */
 public class MasterRegistrationManagerImpl {
-  private static boolean DEBUG = true;
+  private static boolean DEBUG = false;
   private static final Log log = LogFactory.getLog(MasterRegistrationManagerImpl.class);
 
   /**

src/main/java/org/ros/internal/node/server/master/MasterServer.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.server.master;
 
 
@@ -40,7 +24,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.URI;
-import java.net.UnknownHostException;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -55,8 +39,7 @@
  * 
  * @see <a href="http://www.ros.org/wiki/Master">Master documentation</a>
  * 
- * @author [email protected] (Damon Kohler)
- * @author [email protected] (Keith M. Hughes)
+ * @author jg
  */
 public class MasterServer extends RpcServer implements MasterRegistrationListener {
 

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

@@ -1,4 +1,3 @@
-
 package org.ros.internal.node.service;
 
 
@@ -27,7 +26,7 @@
 /**
  * Default implementation of a {@link ServiceClient}.
  * 
- * @author [email protected] (Damon Kohler)
+ * @author jg
  */
 public class DefaultServiceClient<T, S> implements ServiceClient<T, S> {
 
@@ -96,7 +95,7 @@ private DefaultServiceClient(GraphName nodeName, ServiceDeclaration serviceDecla
     // TODO(damonkohler): Support non-persistent connections.
     connectionHeader.addField(ConnectionHeaderFields.PERSISTENT, "1");
     connectionHeader.merge(serviceDeclaration.toConnectionHeader());
-    tcpClientManager = new TcpClientManager(executorService);
+    tcpClientManager = TcpClientManager.getInstance(executorService);
     ServiceClientHandshakeHandler<T, S> serviceClientHandshakeHandler =
         new ServiceClientHandshakeHandler<T, S>(connectionHeader, responseListeners, executorService);
     handshakeLatch = new HandshakeLatch();

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

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

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

@@ -63,7 +63,7 @@ private DefaultSubscriber(NodeIdentifier nodeIdentifier, TopicDeclaration topicD
     this.executorService = executorService;
     incomingMessageQueue = new IncomingMessageQueue<T>(executorService);
     knownPublishers = new HashSet<PublisherIdentifier>();
-    tcpClientManager = new TcpClientManager(executorService);
+    tcpClientManager = TcpClientManager.getInstance(executorService);
     mutex = new Object();
     SubscriberHandshakeHandler<T> subscriberHandshakeHandler =
         new SubscriberHandshakeHandler<T>(toDeclaration().toConnectionHeader(),

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

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

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

@@ -8,6 +8,8 @@
 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;
@@ -28,7 +30,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;
@@ -40,25 +42,40 @@ public SubscriberHandshakeHandler(ConnectionHeader outgoingConnectionHeader,
     if( DEBUG )
     	log.info("subscriberhandshakeHandler ctor:"+this);
   }
-
+  /**
+   * Triggered from BaseClientHandshakeHandler channelRead
+   */
   @Override
   protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandlerContext ctx) {
 	if( DEBUG )
 		log.info("SubscriberHandshakeHandler.onSuccess:"+ctx+" "+incomingConnectionHeader);
     ChannelPipeline pipeline = ctx.pipeline();
-    pipeline.remove(SubscriberHandshakeHandler.this);
+    pipeline.remove(getName());
     NamedChannelHandler namedChannelHandler = incomingMessageQueue.getMessageReceiver();
     pipeline.addLast(namedChannelHandler.getName(), namedChannelHandler);
     String latching = incomingConnectionHeader.getField(ConnectionHeaderFields.LATCHING);
     if (latching != null && latching.equals("1")) {
       incomingMessageQueue.setLatchMode(true);
     }
-    ctx.setReady(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); 
   }
-
+  /**
+   * Triggered from BaseClientHandshakeHandler
+   */
   @Override
   protected void onFailure(String errorMessage, ChannelHandlerContext ctx) throws IOException {
     log.info("Subscriber handshake failed: " + errorMessage);
+    ctx.setReady(false);
     ctx.close();
   }
 

src/main/java/org/ros/internal/system/Utility.java

@@ -20,7 +20,7 @@
  *
  */
 public class Utility {
-	 private static boolean DEBUG = true;
+	 private static boolean DEBUG = false;
 	 private static final Log log = LogFactory.getLog(Utility.class);
 
 	 public static <T> void serialize(T value, ByteBuffer buffer) {