- NioConnection now sends and reads cookie as well; TCP_NIO2 is now a…

…ble to talk to TCP (https://issues.jboss.org/browse/JGRP-1952)

- Added TP to jg-protocols.xml

conf/jg-protocol-ids.xml

@@ -68,6 +68,7 @@
     <class id="72" name="org.jgroups.protocols.FD_HOST"/>
     <class id="73" name="org.jgroups.protocols.ABP"/>
     <class id="74" name="org.jgroups.protocols.TCP_NIO2"/>
+    <class id="75" name="org.jgroups.protocols.TP"/>
 
     <!-- IDs reserved for building blocks -->
     <class id="200" name="org.jgroups.blocks.RequestCorrelator"/> <!-- ID should be the same as Global.BLOCKS_START_ID -->

src/org/jgroups/blocks/cs/Connection.java

@@ -9,14 +9,18 @@
 /**
  * Represents a connection to a peer
  */
-public interface Connection extends Closeable {
-    boolean isOpen();
-    boolean isConnected();
-    Address localAddress();
-    Address peerAddress();
-    boolean isExpired(long millis);
-    void    connect(Address dest) throws Exception;
-    void    start() throws Exception;
-    void    send(byte[] buf, int offset, int length) throws Exception;
-    void    send(ByteBuffer buf) throws Exception;
+public abstract class Connection implements Closeable {
+    protected static final byte[] cookie= { 'b', 'e', 'l', 'a' };
+    protected Address             peer_addr;    // address of the 'other end' of the connection
+    protected long                last_access;  // timestamp of the last access to this connection (read or write)
+
+    abstract public boolean isOpen();
+    abstract public boolean isConnected();
+    abstract public Address localAddress();
+    abstract public Address peerAddress();
+    abstract public boolean isExpired(long millis);
+    abstract public void    connect(Address dest) throws Exception;
+    abstract public void    start() throws Exception;
+    abstract public void    send(byte[] buf, int offset, int length) throws Exception;
+    abstract public void    send(ByteBuffer buf) throws Exception;
 }

src/org/jgroups/blocks/cs/NioConnection.java

@@ -16,6 +16,7 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -25,11 +26,9 @@
  * @author Bela Ban
  * @since  3.6.5
  */
-public class NioConnection implements Connection {
+public class NioConnection extends Connection {
     protected SocketChannel       channel;      // the channel to the peer
     protected SelectionKey        key;
-    protected Address             peer_addr;    // address of the 'other end' of the connection
-    protected long                last_access;  // timestamp of the last access to this connection (read or write)
     protected final NioBaseServer server;
 
     protected final Buffers       send_buf;     // send messages via gathering writes
@@ -39,7 +38,8 @@ public class NioConnection implements Connection {
     protected final Lock          send_lock=new ReentrantLock(); // serialize send()
 
     // creates an array of 2: length buffer (for reading the length of the following data buffer) and data buffer
-    protected Buffers             recv_buf=new Buffers(2).add(ByteBuffer.allocate(Global.INT_SIZE), null);
+    // protected Buffers             recv_buf=new Buffers(2).add(ByteBuffer.allocate(Global.INT_SIZE), null);
+    protected Buffers             recv_buf=new Buffers(4).add(ByteBuffer.allocate(cookie.length));
     protected Reader              reader=new Reader(); // manages the thread which receives messages
     protected long                reader_idle_time=20000; // number of ms a reader can be idle (no msgs) until it terminates
 
@@ -169,10 +169,14 @@ public void send(byte[] buf, int offset, int length) throws Exception {
      */
     @Override
     public void send(ByteBuffer buf) throws Exception {
+        send(buf, true);
+    }
+
+
+    public void send() throws Exception {
         send_lock.lock();
         try {
-            // makeLengthBuffer() reuses the same pre-allocated buffer and copies it only if the write didn't complete
-            boolean success=send_buf.add(makeLengthBuffer(buf), buf).write(channel);
+            boolean success=send_buf.write(channel);
             writeInterest(!success);
             if(success)
                 updateLastAccessed();
@@ -188,9 +192,20 @@ public void send(ByteBuffer buf) throws Exception {
     }
 
 
-    public void send() throws Exception {
+
+    /** Read the length first, then the actual data. This method is not reentrant and access must be synchronized */
+    public void receive() throws Exception {
+        reader.receive();
+    }
+
+    protected void send(ByteBuffer buf, boolean send_length) throws Exception {
         send_lock.lock();
         try {
+            // makeLengthBuffer() reuses the same pre-allocated buffer and copies it only if the write didn't complete
+            if(send_length)
+                send_buf.add(makeLengthBuffer(buf), buf);
+            else
+                send_buf.add(buf);
             boolean success=send_buf.write(channel);
             writeInterest(!success);
             if(success)
@@ -206,25 +221,20 @@ public void send() throws Exception {
         }
     }
 
-
-
-    /** Read the length first, then the actual data. This method is not reentrant and access must be synchronized */
-    public void receive() throws Exception {
-        reader.receive();
-    }
-
-
     protected boolean _receive(boolean update) throws Exception {
         ByteBuffer msg;
         Receiver   receiver=server.receiver();
 
-        if((msg=recv_buf.readLengthAndData(channel)) == null)
-            return false;
         if(peer_addr == null && server.usePeerConnections()) {
-            peer_addr=readPeerAddress(msg);
-            server.addConnection(peer_addr, this);
-            return true;
+            if((peer_addr=readPeerAddress()) != null) {
+                recv_buf=new Buffers(2).add(ByteBuffer.allocate(Global.INT_SIZE), null);
+                server.addConnection(peer_addr, this);
+                return true;
+            }
         }
+
+        if((msg=recv_buf.readLengthAndData(channel)) == null)
+            return false;
         if(receiver != null)
             receiver.receive(peer_addr, msg);
         if(update)
@@ -311,10 +321,12 @@ protected void setSocketParameters(Socket client_sock) throws SocketException {
     protected void sendLocalAddress(Address local_addr) throws Exception {
         try {
             ByteArrayDataOutputStream out=new ByteArrayDataOutputStream();
+            out.write(cookie, 0, cookie.length);
             out.writeShort(Version.version);
+            out.writeShort(local_addr.size()); // address size
             local_addr.writeTo(out);
             ByteBuffer buf=out.getByteBuffer();
-            send(buf);
+            send(buf, false);
             updateLastAccessed();
         }
         catch(Exception ex) {
@@ -323,18 +335,52 @@ protected void sendLocalAddress(Address local_addr) throws Exception {
         }
     }
 
-    protected Address readPeerAddress(ByteBuffer buf) throws Exception {
-        ByteArrayDataInputStream in=new ByteArrayDataInputStream(buf);
-        short version=in.readShort(); // version
-        if(!Version.isBinaryCompatible(version))
-            throw new IOException("packet from " + channel.getRemoteAddress() + " has different version (" + Version.print(version) +
-                                    ") from ours (" + Version.printVersion() + "); discarding it");
-        Address client_peer_addr=new IpAddress();
-        client_peer_addr.readFrom(in);
-        updateLastAccessed();
-        return client_peer_addr;
+    protected Address readPeerAddress() throws Exception {
+        while(recv_buf.read(channel)) {
+            int current_position=recv_buf.position()-1;
+            ByteBuffer buf=recv_buf.get(current_position);
+            if(buf == null)
+                return null;
+            buf.flip();
+            switch(current_position) {
+                case 0:      // cookie
+                    byte[] cookie_buf=getBuffer(buf);
+                    if(!Arrays.equals(cookie, cookie_buf))
+                        throw new IllegalStateException("BaseServer.NioConnection.readPeerAddress(): cookie read by "
+                                                          + server.localAddress() + " does not match own cookie; terminating connection");
+                    recv_buf.add(ByteBuffer.allocate(Global.SHORT_SIZE));
+                    break;
+                case 1:      // version
+                    short version=buf.getShort();
+                    if(!Version.isBinaryCompatible(version))
+                        throw new IOException("packet from " + channel.getRemoteAddress() + " has different version (" + Version.print(version) +
+                                                ") from ours (" + Version.printVersion() + "); discarding it");
+                    recv_buf.add(ByteBuffer.allocate(Global.SHORT_SIZE));
+                    break;
+                case 2:      // length of address
+                    short addr_len=buf.getShort();
+                    recv_buf.add(ByteBuffer.allocate(addr_len));
+                    break;
+                case 3:      // address
+                    byte[] addr_buf=getBuffer(buf);
+                    ByteArrayDataInputStream in=new ByteArrayDataInputStream(addr_buf);
+                    IpAddress addr=new IpAddress();
+                    addr.readFrom(in);
+                    return addr;
+                default:
+                    throw new IllegalStateException(String.format("position %d is invalid", recv_buf.position()));
+            }
+        }
+        return null;
+    }
+
+    protected static byte[] getBuffer(final ByteBuffer buf) {
+        byte[] retval=new byte[buf.limit()];
+        buf.get(retval, buf.position(), buf.limit());
+        return retval;
     }
 
+
     protected static ByteBuffer makeLengthBuffer(ByteBuffer buf) {
         return (ByteBuffer)ByteBuffer.allocate(Global.INT_SIZE).putInt(buf.remaining()).clear();
     }

src/org/jgroups/blocks/cs/TcpConnection.java

@@ -10,6 +10,7 @@
 import java.io.*;
 import java.net.*;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -21,15 +22,12 @@
  * @author Bela Ban
  * @since  3.6.5
  */
-public class TcpConnection implements Connection {
+public class TcpConnection extends Connection {
     protected final Socket           sock; // socket to/from peer (result of srv_sock.accept() or new Socket())
     protected final ReentrantLock    send_lock=new ReentrantLock(); // serialize send()
-    protected static final byte[]    cookie= { 'b', 'e', 'l', 'a' };
     protected static final Buffer    termination=new Buffer(cookie);
     protected DataOutputStream       out;
     protected DataInputStream        in;
-    protected Address                peer_addr; // address of the 'other end' of the connection
-    protected long                   last_access;
     protected volatile Sender        sender;
     protected volatile Receiver      receiver;
     protected final TcpBaseServer    server;
@@ -231,6 +229,7 @@ protected void sendLocalAddress(Address local_addr) throws Exception {
 
             // write the version
             out.writeShort(Version.version);
+            out.writeShort(local_addr.size()); // address size
             local_addr.writeTo(out);
             out.flush(); // needed ?
             updateLastAccessed();
@@ -253,7 +252,7 @@ protected Address readPeerAddress(Socket client_sock) throws Exception {
             // read the cookie first
             byte[] input_cookie=new byte[cookie.length];
             in.readFully(input_cookie, 0, input_cookie.length);
-            if(!matchCookie(input_cookie))
+            if(!Arrays.equals(cookie, input_cookie))
                 throw new SocketException("BaseServer.TcpConnection.readPeerAddress(): cookie read by "
                                             + server.localAddress() + " does not match own cookie; terminating connection");
             // then read the version
@@ -262,6 +261,8 @@ protected Address readPeerAddress(Socket client_sock) throws Exception {
                 throw new IOException("packet from " + client_sock.getInetAddress() + ":" + client_sock.getPort() +
                                         " has different version (" + Version.print(version) +
                                         ") from ours (" + Version.printVersion() + "); discarding it");
+            short addr_len=in.readShort(); // only needed by NioConnection
+
             Address client_peer_addr=new IpAddress();
             client_peer_addr.readFrom(in);
             updateLastAccessed();
@@ -273,12 +274,6 @@ protected Address readPeerAddress(Socket client_sock) throws Exception {
     }
 
 
-    protected static boolean matchCookie(byte[] input) {
-        if(input == null || input.length < cookie.length) return false;
-        for(int i=0; i < cookie.length; i++)
-            if(cookie[i] != input[i]) return false;
-        return true;
-    }
 
     protected class Receiver implements Runnable {
         protected final Thread     recv;

src/org/jgroups/conf/ClassConfigurator.java

@@ -192,9 +192,7 @@ public static short getMagicNumber(Class clazz) {
 
     public static short getProtocolId(Class protocol) {
         Short retval=protocol_ids.get(protocol);
-        if(retval != null)
-            return retval;
-        return 0;
+        return retval != null? retval : 0;
     }
 
 

src/org/jgroups/nio/Buffers.java

@@ -153,7 +153,7 @@ public ByteBuffer readLengthAndData(SocketChannel ch) throws Exception {
      * {@link ByteBuffer#clear()} has to be called (all buffers have their position == limit on a successfull read).
      */
     public boolean read(SocketChannel ch) throws Exception {
-        long bytes=ch.read(bufs, position, limit);
+        long bytes=ch.read(bufs, position, limit-position);
         if(bytes == -1)
             throw new EOFException();
         return adjustPosition(false);

src/org/jgroups/nio/MockSocketChannel.java

@@ -48,6 +48,8 @@ public MockSocketChannel bytesToRead(ByteBuffer buf) {
         return this;
     }
 
+    public ByteBuffer bytesToRead() {return bytes_to_read;}
+
     public MockSocketChannel recorder(ByteBuffer buf) {this.recorder=buf; return this;}
     public ByteBuffer        recorder()               {return recorder;}
 

src/org/jgroups/protocols/TP.java

@@ -4,6 +4,7 @@
 import org.jgroups.*;
 import org.jgroups.annotations.*;
 import org.jgroups.blocks.LazyRemovalCache;
+import org.jgroups.conf.ClassConfigurator;
 import org.jgroups.conf.PropertyConverters;
 import org.jgroups.logging.LogFactory;
 import org.jgroups.stack.DiagnosticsHandler;
@@ -1085,7 +1086,8 @@ public void evictLogicalAddressCache(boolean force) {
 
 
     public void init() throws Exception {
-        super.init();
+        this.id=ClassConfigurator.getProtocolId(TP.class);
+        // super.init();
 
         // Create the default thread factory
         if(global_thread_factory == null)