Alpha full traffic but unfiltered subscriber mssages

Author: neocoretechs

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

@@ -143,9 +143,9 @@ public T newMessage() {
 
   @Override
   public void publish(T message) {
-    if (DEBUG) {
-      log.info(String.format("Publishing message %s on topic %s.", message, getTopicName()));
-    }
+    //if (DEBUG) {
+    //  log.info(String.format("Publishing message %s on topic %s.", message, getTopicName()));
+    //}
     outgoingMessageQueue.add(message);
   }
 
@@ -180,7 +180,7 @@ 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.encode();
+    return (ByteBuffer) outgoingConnectionHeader.encode();
   }
 
   /**

src/main/java/org/ros/internal/node/topic/SubscriberHandshake.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;
@@ -24,8 +8,11 @@
 
 /**
  * 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.
  * 
- * @author damonkohler@google.com (Damon Kohler)
+ * @author jg
  */
 public class SubscriberHandshake extends BaseClientHandshake {
 

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

@@ -53,11 +53,12 @@ protected void onSuccess(ConnectionHeader incomingConnectionHeader, ChannelHandl
     if (latching != null && latching.equals("1")) {
       incomingMessageQueue.setLatchMode(true);
     }
+    ctx.setReady(true);
   }
 
   @Override
   protected void onFailure(String errorMessage, ChannelHandlerContext ctx) throws IOException {
-    log.error("Subscriber handshake failed: " + errorMessage);
+    log.info("Subscriber handshake failed: " + errorMessage);
     ctx.close();
   }
 

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

@@ -66,6 +66,7 @@ public void run() {
     } catch (Exception e) {
       // TODO(damonkohler): Retry logic is needed at the RPC layer.
       log.error(e);
+      e.printStackTrace();
     }
   }
 }

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

@@ -13,7 +13,8 @@
 import org.ros.internal.message.field.DirectByteArrayOutputStream;
 
 public class Utility {
-	  private static final Log log = LogFactory.getLog(Utility.class);
+	 private static boolean DEBUG = true;
+	 private static final Log log = LogFactory.getLog(Utility.class);
 	 public static <T> void serialize(T value, ByteBuffer buffer) {
 		    //serializer.serialize((Message) value, buffer);
 			DirectByteArrayOutputStream dbaos = new DirectByteArrayOutputStream();
@@ -22,8 +23,10 @@ public static <T> void serialize(T value, ByteBuffer buffer) {
 				oos = new ObjectOutputStream(dbaos);
 				oos.writeObject(value);
 				oos.flush();
+				buffer.clear();
 				buffer.put(dbaos.getBuf());
 				oos.close();
+				buffer.flip();
 			} catch (IOException e) {
 				e.printStackTrace();
 			}
@@ -33,20 +36,24 @@ public static <T> void serialize(T value, ByteBuffer buffer) {
 	 public static Object deserialize(ByteBuffer buffer) {
 		    //return deserializer.deserialize(buffer);
 			byte[] obuf = buffer.array();
+			if( DEBUG )
+				log.info("Deserialize:"+obuf.length);
 			Object Od = null;
 			try {
 						ObjectInputStream s;
 						ByteArrayInputStream bais = new ByteArrayInputStream(obuf);
-						ReadableByteChannel rbc = Channels.newChannel(bais);
-						s = new ObjectInputStream(Channels.newInputStream(rbc));
+						//ReadableByteChannel rbc = Channels.newChannel(bais);
+						s = new ObjectInputStream(bais/*Channels.newInputStream(rbc)*/);
 						Od = s.readObject();
 						s.close();
 						bais.close();
-						rbc.close();
+						//rbc.close();
 			} catch (IOException ioe) {
 			} catch (ClassNotFoundException cnf) {
 					log.error("Class cannot be deserialized, may have been modified beyond version compatibility");
 			}
+			if( DEBUG )
+				log.info("Deserialize return:"+Od);
 			return  Od;
 
 	}

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

@@ -1,9 +1,5 @@
 package org.ros.internal.transport;
 
-//import org.jboss.netty.channel.ChannelHandlerContext;
-//import org.jboss.netty.channel.ChannelStateEvent;
-//import org.jboss.netty.channel.MessageEvent;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.ros.concurrent.ListenerGroup;
@@ -29,7 +25,10 @@ public BaseClientHandshakeHandler(ClientHandshake clientHandshake, ExecutorServi
     this.clientHandshake = clientHandshake;
     clientHandshakeListeners = new ListenerGroup<ClientHandshakeListener>(executorService);
   }
-
+  /**
+   * Primarily services?
+   * @param clientHandshakeListener
+   */
   public void addListener(ClientHandshakeListener clientHandshakeListener) {
     clientHandshakeListeners.add(clientHandshakeListener);
   }

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

@@ -35,7 +35,7 @@ public interface ChannelHandlerContext {
      * <p>
      * This will result in having the
      * {@link ChannelOutboundHandler#bind(ChannelHandlerContext, SocketAddress, CompletionHandler)} method
-     * called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
+     * called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      */
     AsynchronousSocketChannel bind(SocketAddress localAddress) throws IOException;
@@ -50,7 +50,7 @@ public interface ChannelHandlerContext {
      * <p>
      * This will result in having the
      * {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, CompletionHandler)}
-     * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
+     * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      */
     void connect(SocketAddress remoteAddress);
@@ -61,7 +61,7 @@ public interface ChannelHandlerContext {
      * an error.
      * <p>
      * This will result in having the
-     * {@link ChannelOutboundHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, CompletionHandler)}
+     * {@link ChannelHandler#connect(ChannelHandlerContext, SocketAddress, SocketAddress, CompletionHandler)}
      * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      * @throws IOException 
@@ -75,11 +75,12 @@ public interface ChannelHandlerContext {
      * either because the operation was successful or because of an error.
      * <p>
      * This will result in having the
-     * {@link ChannelOutboundHandler#disconnect(ChannelHandlerContext, CompletionHandler)}
-     * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
+     * {@link ChannelHandler#disconnect(ChannelHandlerContext, CompletionHandler)}
+     * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
+     * @throws IOException 
      */
-    Future<Void> disconnect();
+    void disconnect() throws IOException;
 
     /**
      * Request to close the {@link Channel} and notify the {@link Future} once the operation completes,
@@ -89,8 +90,8 @@ public interface ChannelHandlerContext {
      * After it is closed it is not possible to reuse it again.
      * <p>
      * This will result in having the
-     * {@link ChannelOutboundHandler#close(ChannelHandlerContext, CompletionHandler)}
-     * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
+     * {@link ChannelHandler#close(ChannelHandlerContext, CompletionHandler)}
+     * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      * @throws IOException 
      */
@@ -104,11 +105,24 @@ public interface ChannelHandlerContext {
      * handler can decide to continue reading.  If there's a pending read operation already, this method does nothing.
      * <p>
      * This will result in having the
-     * {@link ChannelOutboundHandler#read(ChannelHandlerContext)}
-     * method called of the next {@link ChannelOutboundHandler} contained in the  {@link ChannelPipeline} of the
+     * {@link ChannelHandler#read(ChannelHandlerContext)}
+     * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
      * {@link Channel}.
      */
     Future<Integer> read(ByteBuffer buf);
+    /**
+     * Request to Read data from the {@link Channel} into the first inbound buffer, triggers an
+     * {@link ChannelHandler#channelRead(ChannelHandlerContext, Object)} event if data was
+     * read, and triggers a
+     * {@link ChannelHandler#channelReadComplete(ChannelHandlerContext) channelReadComplete} event so the
+     * handler can decide to continue reading.  If there's a pending read operation already, this method does nothing.
+     * <p>
+     * This will result in having the
+     * {@link ChannelHandler#read(ChannelHandlerContext)}
+     * method called of the next {@link ChannelHandler} contained in the  {@link ChannelPipeline} of the
+     * {@link Channel}.
+     */
+	void read(ByteBuffer buf,CompletionHandler<Integer, Void> completionHandler);
 
     /**
      * Request to write a message via this {@link ChannelHandlerContext} through the {@link ChannelPipeline}.
@@ -122,22 +136,8 @@ public interface ChannelHandlerContext {
      * This method will not request to actual flush, so be sure to call {@link #flush()}
      * once you want to request to flush all pending data to the actual transport.
      */
-    Future<Void> write(Object msg, CompletionHandler<Integer,Void> handler);
-
-    /**
-     * Request to flush all pending messages via this ChannelOutboundInvoker.
-     */
-    ChannelHandlerContext flush();
-
-    /**
-     * Shortcut for call {@link #write(Object, CompletionHandler)} and {@link #flush()}.
-     */
-    Future<Void> writeAndFlush(Object msg, CompletionHandler<Integer,Void> handler);
+    void write(Object msg, CompletionHandler<Integer,Void> handler);
 
-    /**
-     * Shortcut for call {@link #write(Object)} and {@link #flush()}.
-     */
-    Future<Integer> writeAndFlush(Object msg);
 
     /**
      * Return the assigned {@link ChannelPipeline}
@@ -150,7 +150,15 @@ public interface ChannelHandlerContext {
      */
     AsynchronousChannelGroup getChannelGroup();
 
+    /**
+     * Determine if this channel is ready for processing, it is configured, has a socket
+     * and the communication is sound. If the socket breaks this goes false and no writes are
+     * performed to this channel
+     */
+    boolean isReady();
+    
+    void setReady(boolean ready);
 
-
+    Object getChannelCompletionMutex();
 
 }

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

@@ -29,6 +29,8 @@ public class ChannelHandlerContextImpl implements ChannelHandlerContext {
 	Executor executor;
 	AsynchronousSocketChannel channel;
 	ChannelPipeline pipeline;
+	boolean ready = false;
+	Object mutex = new Object();
 
 
 	public ChannelHandlerContextImpl(AsynchronousChannelGroup grp, AsynchronousSocketChannel ch, Executor exc) {
@@ -72,8 +74,8 @@ public void connect(SocketAddress remoteAddress, SocketAddress localAddress) thr
 
 
 	@Override
-	public Future<Void> disconnect() {
-		throw new RuntimeException("nope..");
+	public void disconnect() throws IOException {
+		channel.close();
 	}
 
 	@Override
@@ -93,26 +95,15 @@ public Future<Integer> write(Object msg) {
 	}
 
 	@Override
-	public Future<Void> write(Object msg,
-			CompletionHandler<Integer, Void> handler) {
-		throw new RuntimeException("nope..");
+	public void write(Object msg, CompletionHandler<Integer, Void> handler) {
+		channel.write((ByteBuffer)msg, null, handler);
 	}
 
 	@Override
-	public ChannelHandlerContext flush() {
-		throw new RuntimeException("nope..");
+	public void read(ByteBuffer buf, CompletionHandler<Integer, Void> handler) {
+		channel.read(buf, null, handler);
 	}
 
-	@Override
-	public Future<Void> writeAndFlush(Object msg,
-			CompletionHandler<Integer, Void> handler) {
-		throw new RuntimeException("nope..");
-	}
-
-	@Override
-	public Future<Integer> writeAndFlush(Object msg) {
-		return channel.write((ByteBuffer) msg);
-	}
 
 	@Override
 	public ChannelPipeline pipeline() {
@@ -131,8 +122,19 @@ public Channel channel() {
 
 	@Override
 	public String toString() {
-		return new String("ChannelHandlerContext:"+channel+" "+channelGroup+" "+executor+" "+pipeline);
+		return new String("ChannelHandlerContext:"+channel+" "+channelGroup+" "+executor+" "+pipeline+" ready:"+ready);
 	}
 
+	@Override
+	public boolean isReady() {
+		return ready;
+	}
 
+	public void setReady(boolean ready) { this.ready = ready;}
+
+	/**
+	 * Object to synchronize read and write completion for the channel in this context, since we will have
+	 * multiple outbound writers accessing the same channel
+	 */
+	public Object getChannelCompletionMutex() { return mutex; }
 }