Removing Netty, streamline

Author: neocoretechs

src/main/java/build.xml

@@ -26,7 +26,7 @@
 
   <!-- Compiles the java code (including the usage of library for JUnit -->
   <target name="compile" depends="makedir">
-    <javac srcdir="${src.dir}" destdir="${build.dir}" classpath="${lib.dir}/netty.Final.jar:${lib.dir}/commons-pool-1.6.jar:${lib.dir}/commons-lang-2.6.jar:${lib.dir}/ws-commons-util-1.0.1.jar:${lib.dir}/en_us.jar:${lib.dir}/commons-digester-1.8.1.jar:${lib.dir}/commons-logging-1.1.2.jar:${lib.dir}/commons-io-2.4.jar:${lib.dir}/RosBase.jar:${lib.dir}/commons-net-1.4.1.jar:${lib.dir}/RosMsgs.jar">
+    <javac srcdir="${src.dir}" destdir="${build.dir}" classpath="${lib.dir}/commons-pool-1.6.jar:${lib.dir}/commons-lang-2.6.jar:${lib.dir}/ws-commons-util-1.0.1.jar:${lib.dir}/en_us.jar:${lib.dir}/commons-digester-1.8.1.jar:${lib.dir}/commons-logging-1.1.2.jar:${lib.dir}/commons-io-2.4.jar:${lib.dir}/RosBase.jar:${lib.dir}/commons-net-1.4.1.jar:${lib.dir}/RosMsgs.jar">
     </javac>
 
   </target>

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

@@ -1,22 +1,5 @@
-/*
- * 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;
 
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.ros.Parameters;
@@ -66,8 +49,6 @@
 import org.ros.time.ClockTopicTimeProvider;
 import org.ros.time.TimeProvider;
 
-import io.netty.channel.nio.NioEventLoop;
-
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.URI;
@@ -79,14 +60,10 @@
 /**
  * The default implementation of a {@link Node}.
  * 
- * @author ethan.rublee@gmail.com (Ethan Rublee)
- * @author kwc@willowgarage.com (Ken Conley)
- * @author damonkohler@google.com (Damon Kohler)
  */
 public class DefaultNode implements ConnectedNode {
   private static final boolean DEBUG = true;
-  private static final Log log = LogFactory.getLog(TcpRosServer.class);
-  // TODO(damonkohler): Move to NodeConfiguration.
+  private static final Log log = LogFactory.getLog(DefaultNode.class);
   /**
    * The maximum delay before shutdown will begin even if all
    * {@link NodeListener}s have not yet returned from their
@@ -97,7 +74,7 @@ public class DefaultNode implements ConnectedNode {
 
   private final NodeConfiguration nodeConfiguration;
   private final ListenerGroup<NodeListener> nodeListeners;
-  private final /*ScheduledExecutorService*/NioEventLoop scheduledExecutorService;
+  private final ScheduledExecutorService scheduledExecutorService;
   private final InetSocketAddress masterUri;
   private final MasterClient masterClient;
   private final TopicParticipantManager topicParticipantManager;
@@ -126,7 +103,7 @@ public class DefaultNode implements ConnectedNode {
    *          to this {@link Node} before it starts
    */
   public DefaultNode(NodeConfiguration nodeConfiguration, Collection<NodeListener> nodeListeners,
-      /*ScheduledExecutorService*/NioEventLoop scheduledExecutorService) {
+      ScheduledExecutorService scheduledExecutorService) {
     this.nodeConfiguration = NodeConfiguration.copyOf(nodeConfiguration);
     this.nodeListeners = new ListenerGroup<NodeListener>(scheduledExecutorService);
     this.nodeListeners.addAll(nodeListeners);
@@ -199,14 +176,18 @@ private void start() {
     // During startup, we wait for 1) the RosoutLogger and 2) the TimeProvider.
     final CountDownLatch latch = new CountDownLatch(2);
 
-    rlog = new RosoutLogger(this);
-    rlog.getPublisher().addListener(new DefaultPublisherListener<rosgraph_msgs.Log>() {
-      @Override
-      public void onMasterRegistrationSuccess(Publisher<rosgraph_msgs.Log> registrant) {
-        latch.countDown();
-      }
-    });
-
+    try {
+		rlog = new RosoutLogger(this);
+	    rlog.getPublisher().addListener(new DefaultPublisherListener<rosgraph_msgs.Log>() {
+	        @Override
+	        public void onMasterRegistrationSuccess(Publisher<rosgraph_msgs.Log> registrant) {
+	          latch.countDown();
+	        }
+	    });
+	} catch (IOException e1) {
+		log.error("The Ros Logger has failed to start.",e1);
+		throw new RuntimeException(e1);
+	}
     boolean useSimTime = false;
     try {
       useSimTime =
@@ -256,7 +237,11 @@ public <T> Publisher<T> newPublisher(GraphName topicName, String messageType) {
         nodeConfiguration.getTopicDescriptionFactory().newFromType(messageType);
     TopicDeclaration topicDeclaration =
         TopicDeclaration.newFromTopicName(resolvedTopicName, topicDescription);
-    return publisherFactory.newOrExisting(topicDeclaration);
+    Publisher<T> publisher = null;
+    try {
+     publisher = publisherFactory.newOrExisting(topicDeclaration, slaveServer.getSubscribers());
+    } catch(IOException e) { throw new RosRuntimeException(e); }
+    return publisher;
   }
 
   @Override
@@ -271,7 +256,10 @@ public <T> Subscriber<T> newSubscriber(GraphName topicName, String messageType)
         nodeConfiguration.getTopicDescriptionFactory().newFromType(messageType);
     TopicDeclaration topicDeclaration =
         TopicDeclaration.newFromTopicName(resolvedTopicName, topicDescription);
-    Subscriber<T> subscriber = subscriberFactory.newOrExisting(topicDeclaration);
+    Subscriber<T> subscriber = null;
+    try {
+    	subscriber = subscriberFactory.newOrExisting(topicDeclaration);
+    } catch(IOException e) { throw new RosRuntimeException(e); }
     return subscriber;
   }
 
@@ -328,8 +316,7 @@ public InetSocketAddress lookupServiceUri(String serviceName) {
   }
 
   @Override
-  public <T, S> ServiceClient<T, S> newServiceClient(GraphName serviceName, String serviceType)
-      throws ServiceNotFoundException {
+  public <T, S> ServiceClient<T, S> newServiceClient(GraphName serviceName, String serviceType) throws Exception {
     GraphName resolvedServiceName = resolveName(serviceName);
     InetSocketAddress uri = lookupServiceUri(resolvedServiceName);
     if (uri == null) {
@@ -344,8 +331,7 @@ public <T, S> ServiceClient<T, S> newServiceClient(GraphName serviceName, String
   }
 
   @Override
-  public <T, S> ServiceClient<T, S> newServiceClient(String serviceName, String serviceType)
-      throws ServiceNotFoundException {
+  public <T, S> ServiceClient<T, S> newServiceClient(String serviceName, String serviceType) throws Exception {
     return newServiceClient(GraphName.of(serviceName), serviceType);
   }
 

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

@@ -21,6 +21,7 @@
 import org.ros.Topics;
 import org.ros.node.topic.Publisher;
 
+import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 
@@ -36,7 +37,7 @@ class RosoutLogger implements Log {
   private final Publisher<rosgraph_msgs.Log> publisher;
   private final Log log;
 
-  public RosoutLogger(DefaultNode defaultNode) {
+  public RosoutLogger(DefaultNode defaultNode) throws IOException {
     this.defaultNode = defaultNode;
     publisher = defaultNode.newPublisher(Topics.ROSOUT, rosgraph_msgs.Log._TYPE);
     log = LogFactory.getLog(defaultNode.getName().toString());

src/main/java/org/ros/internal/node/server/SlaveServer.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;
 
 import org.ros.address.AdvertiseAddress;
@@ -28,15 +12,13 @@
 import org.ros.internal.node.topic.TopicDeclaration;
 import org.ros.internal.node.topic.TopicParticipantManager;
 import org.ros.internal.system.Process;
+import org.ros.internal.transport.ChannelHandlerContext;
 import org.ros.internal.transport.ProtocolDescription;
 import org.ros.internal.transport.ProtocolNames;
 import org.ros.internal.transport.tcp.TcpRosProtocolDescription;
 import org.ros.internal.transport.tcp.TcpRosServer;
 import org.ros.namespace.GraphName;
 
-import io.netty.channel.EventLoop;
-import io.netty.channel.nio.NioEventLoopGroup;
-
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
@@ -45,7 +27,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 
 /**
- * @author damonkohler@google.com (Damon Kohler)
+ * @author jg
  */
 public class SlaveServer extends RpcServer {
 
@@ -71,15 +53,21 @@ public SlaveServer(GraphName nodeName, BindAddress tcpRosBindAddress,
 	} catch (ClassNotFoundException e) {
 		throw new IOException(e);
 	}
-    EventLoop exec = new NioEventLoopGroup().next();
+   
     this.tcpRosServer =
-        new TcpRosServer(tcpRosBindAddress, tcpRosAdvertiseAddress, topicParticipantManager, serviceManager, exec/*utorService*/);
+        new TcpRosServer(tcpRosBindAddress, tcpRosAdvertiseAddress, topicParticipantManager, serviceManager, executorService);
   }
 
   public AdvertiseAddress getTcpRosAdvertiseAddress() {
     return tcpRosServer.getAdvertiseAddress();
   }
-
+  /**
+   * Return the ChannelHandlerContext array of subscribers
+   * @return
+   */
+  public List<ChannelHandlerContext> getSubscribers() {
+	  return tcpRosServer.getSubscribers();
+  }
   /**
    * Start the RPC server. This start() routine requires that the
    * {@link TcpRosServer} is initialized first so that the slave server returns

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

@@ -2,7 +2,6 @@
 package org.ros.internal.node.service;
 
 
-//import org.jboss.netty.buffer.ChannelBuffer;
 import org.ros.exception.RosRuntimeException;
 import org.ros.internal.message.MessageBufferPool;
 import org.ros.internal.system.Utility;
@@ -16,11 +15,9 @@
 import org.ros.node.service.ServiceClient;
 import org.ros.node.service.ServiceResponseListener;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.util.concurrent.EventExecutor;
-
+import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.net.URI;
+import java.nio.ByteBuffer;
 import java.util.LinkedList;
 import java.util.Queue;
 import java.util.concurrent.CountDownLatch;
@@ -84,12 +81,12 @@ public void reset() {
   public static <S, T> DefaultServiceClient<S, T> newDefault(GraphName nodeName,
       ServiceDeclaration serviceDeclaration,
       MessageFactory messageFactory,
-      ScheduledExecutorService executorService) {
+      ScheduledExecutorService executorService) throws IOException {
     return new DefaultServiceClient<S, T>(nodeName, serviceDeclaration, messageFactory, executorService);
   }
 
   private DefaultServiceClient(GraphName nodeName, ServiceDeclaration serviceDeclaration,
-      MessageFactory messageFactory, ScheduledExecutorService executorService) {
+      MessageFactory messageFactory, ScheduledExecutorService executorService) throws IOException {
     this.serviceDeclaration = serviceDeclaration;
     this.messageFactory = messageFactory;
     messageBufferPool = new MessageBufferPool();
@@ -99,7 +96,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((EventExecutor) executorService);
+    tcpClientManager = new TcpClientManager(executorService);
     ServiceClientHandshakeHandler<T, S> serviceClientHandshakeHandler =
         new ServiceClientHandshakeHandler<T, S>(connectionHeader, responseListeners, executorService);
     handshakeLatch = new HandshakeLatch();
@@ -108,12 +105,16 @@ private DefaultServiceClient(GraphName nodeName, ServiceDeclaration serviceDecla
   }
 
   @Override
-  public void connect(InetSocketAddress inetSocketAddress) {
+  public void connect(InetSocketAddress inetSocketAddress) throws Exception {
     assert(inetSocketAddress != null) : "Address must be specified.";
     //assert(inetSocketAddress.getScheme().equals("rosrpc")) : "Invalid service URI.";
     assert(tcpClient == null) : "Already connected once.";
     handshakeLatch.reset();
-    tcpClient = tcpClientManager.connect(toString(), inetSocketAddress);
+    try {
+		tcpClient = tcpClientManager.connect(toString(), inetSocketAddress);
+	} catch (IOException e1) {
+		throw new RosRuntimeException("TcpClient failed to connect to remote "+toString()+" "+inetSocketAddress, e1);
+	}
     try {
       if (!handshakeLatch.await(1, TimeUnit.SECONDS)) {
         throw new RosRuntimeException(handshakeLatch.getErrorMessage());
@@ -131,11 +132,11 @@ public void shutdown() {
 
   @Override
   public void call(T request, ServiceResponseListener<S> listener) {
-    ByteBuf buffer = messageBufferPool.acquire();
+    ByteBuffer buffer = messageBufferPool.acquire();
     // serialize request to buffer
     Utility.serialize(request, buffer);
     responseListeners.add(listener);
-    tcpClient.write(buffer).awaitUninterruptibly();
+    tcpClient.write(buffer);
     messageBufferPool.release(buffer);
   }
 

src/main/java/org/ros/internal/node/service/DefaultServiceServer.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.service;
 
 
@@ -26,6 +10,7 @@
 import org.ros.concurrent.SignalRunnable;
 import org.ros.internal.message.service.ServiceDescription;
 import org.ros.internal.node.topic.DefaultPublisher;
+import org.ros.internal.transport.ChannelHandler;
 import org.ros.internal.transport.ConnectionHeader;
 import org.ros.internal.transport.ConnectionHeaderFields;
 import org.ros.message.MessageFactory;
@@ -35,11 +20,9 @@
 import org.ros.node.service.ServiceServer;
 import org.ros.node.service.ServiceServerListener;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelHandler;
-
 import java.net.InetSocketAddress;
 import java.net.URI;
+import java.nio.ByteBuffer;
 import java.util.concurrent.ScheduledExecutorService;
 
 /**
@@ -92,7 +75,7 @@ public void onMasterUnregistrationFailure(ServiceServer<T, S> registrant) {
     });
   }
 
-  public ByteBuf finishHandshake(ConnectionHeader incomingConnectionHeader) {
+  public ByteBuffer finishHandshake(ConnectionHeader incomingConnectionHeader) {
     if (DEBUG) {
       log.info("Client handshake header: " + incomingConnectionHeader);
     }