- TP: moved version checking from handleSingleMessage() and handleMes…

…sageBatch() into receive()

- TP implements both receive(InputStream) and receive(byte[]) [https://issues.jboss.org/browse/JGRP-2165]
- BaseServer discards length bytes if no receiver callback has been installed
- TcpConnection now doesn't create intermediate buffer, but passes the input stream directly to the receive() callback
- GossipRouter/RouterStub now implement both receive() callbacks
- PubClient/PubServer now implement both receive() callbacks
- RoundTrip.receive() never used correct offsets to read data

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

@@ -12,6 +12,7 @@
 import org.jgroups.util.*;
 
 import java.io.Closeable;
+import java.io.DataInput;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
@@ -165,6 +166,15 @@ public void receive(Address sender, ByteBuffer buf) {
             this.receiver.receive(sender, buf);
     }
 
+    public void receive(Address sender, DataInput in, int len) throws Exception {
+        if(this.receiver != null)
+            this.receiver.receive(sender, in);
+        else {
+            // discard len bytes (in.skip() is not guaranteed to discard *all* len bytes)
+            byte[] buf=new byte[len];
+            in.readFully(buf, 0, len);
+        }
+    }
 
 
     public void send(Address dest, byte[] data, int offset, int length) throws Exception {

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

@@ -2,6 +2,7 @@
 
 import org.jgroups.Address;
 
+import java.io.DataInput;
 import java.nio.ByteBuffer;
 
 /**
@@ -31,4 +32,6 @@ public interface Receiver {
      *            Note that buf could be a direct ByteBuffer.
      */
     void receive(Address sender, ByteBuffer buf); // should be a default method in Java 8
+
+    void receive(Address sender, DataInput in) throws Exception;
 }

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

@@ -3,6 +3,7 @@
 import org.jgroups.Address;
 import org.jgroups.util.Util;
 
+import java.io.DataInput;
 import java.nio.ByteBuffer;
 
 /**
@@ -25,5 +26,7 @@ public void receive(Address sender, ByteBuffer buf) {
         Util.bufferToArray(sender, buf, this);
     }
 
+    public void receive(Address sender, DataInput in) throws Exception {
 
+    }
 }

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

@@ -286,12 +286,9 @@ public void run() {
             Throwable t=null;
             while(canRun()) {
                 try {
-                    int len=in.readInt();
-                    if(buffer == null || buffer.length < len)
-                        buffer=new byte[len];
-                    in.readFully(buffer, 0, len);
+                    int len=in.readInt(); // needed to read messages from TCP_NIO2
+                    server.receive(peer_addr, in, len);
                     updateLastAccessed();
-                    server.receive(peer_addr, buffer, 0, len);
                 }
                 catch(OutOfMemoryError mem_ex) {
                     t=mem_ex;

src/org/jgroups/demos/PubClient.java

@@ -1,10 +1,13 @@
 package org.jgroups.demos;
 
 import org.jgroups.Address;
+import org.jgroups.Global;
 import org.jgroups.blocks.cs.*;
+import org.jgroups.util.Bits;
 import org.jgroups.util.Util;
 
 import java.io.BufferedInputStream;
+import java.io.DataInput;
 import java.io.InputStream;
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
@@ -28,16 +31,26 @@ public PubClient(String name) {
 
     @Override
     public void receive(Address sender, ByteBuffer buf) {
-        String msg=new String(buf.array(), buf.arrayOffset(), buf.limit());
+        byte[] buffer=buf.array();
+        int len=Bits.readInt(buffer, buf.arrayOffset());
+        String msg=new String(buffer, buf.arrayOffset()+Global.INT_SIZE, len);
         System.out.printf("-- %s\n", msg);
     }
 
     @Override
     public void receive(Address sender, byte[] buf, int offset, int length) {
-        String msg=new String(buf, offset, length);
+        int len=Bits.readInt(buf, offset);
+        String msg=new String(buf, offset+Global.INT_SIZE, len);
         System.out.printf("-- %s\n", msg);
     }
 
+    @Override public void receive(Address sender, DataInput in) throws Exception {
+        int len=in.readInt();
+        byte[] buf=new byte[len];
+        in.readFully(buf);
+        String msg=new String(buf, 0, buf.length);
+        System.out.printf("-- %s\n", msg);
+    }
 
     @Override
     public void connectionClosed(Connection conn, String cause) {
@@ -58,8 +71,7 @@ protected void start(InetAddress srv_addr, int srv_port, boolean nio) throws Exc
         client.receiver(this);
         client.addConnectionListener(this);
         client.start();
-        byte[] buf=String.format("%s joined\n", name).getBytes();
-        ((Client)client).send(buf, 0, buf.length);
+        send(String.format("%s joined", name));
         eventLoop();
         client.stop();
     }
@@ -75,8 +87,7 @@ protected void eventLoop() {
                 if(line.startsWith("quit") || line.startsWith("exit")) {
                     break;
                 }
-                byte[] buf=String.format("%s: %s\n", name, line).getBytes();
-                ((Client)client).send(buf, 0, buf.length);
+                send(String.format("%s: %s", name, line));
             }
             catch(Exception e) {
                 e.printStackTrace();
@@ -85,6 +96,14 @@ protected void eventLoop() {
         }
     }
 
+    protected void send(String str) throws Exception {
+        byte[] buf=str.getBytes();
+        byte[] data=new byte[Global.INT_SIZE + buf.length];
+        Bits.writeInt(buf.length, data, 0);
+        System.arraycopy(buf, 0, data, Global.INT_SIZE, buf.length);
+        ((Client)client).send(data, 0, data.length);
+    }
+
     public static void main(String[] args) throws Exception {
         InetAddress server_addr=InetAddress.getLocalHost();
         int         server_port=7500;

src/org/jgroups/demos/PubServer.java

@@ -1,13 +1,16 @@
 package org.jgroups.demos;
 
 import org.jgroups.Address;
+import org.jgroups.Global;
 import org.jgroups.blocks.cs.*;
 import org.jgroups.jmx.JmxConfigurator;
 import org.jgroups.logging.Log;
 import org.jgroups.logging.LogFactory;
 import org.jgroups.stack.IpAddress;
+import org.jgroups.util.Bits;
 import org.jgroups.util.Util;
 
+import java.io.DataInput;
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
 
@@ -52,6 +55,14 @@ public void receive(Address sender, byte[] buf, int offset, int length) {
         }
     }
 
+    public void receive(Address sender, DataInput in) throws Exception {
+        int len=in.readInt();
+        byte[] buf=new byte[len + Global.INT_SIZE];
+        Bits.writeInt(len, buf, 0);
+        in.readFully(buf, Global.INT_SIZE, len);
+        server.send(null, buf, 0, buf.length);
+    }
+
     public static void main(String[] args) throws Exception {
         int         port=7500;
         InetAddress bind_addr=null;

src/org/jgroups/protocols/BasicTCP.java

@@ -77,8 +77,6 @@ protected BasicTCP() {
     public void     setReaperInterval(long interval) {this.reaper_interval=interval;}
     public long     getConnExpireTime()              {return conn_expire_time;}
     public void     setConnExpireTime(long time)     {this.conn_expire_time=time;}
-    // public boolean  getReuseAddress()                {return this.reuse_addr;}
-    // public BasicTCP setReuseAddress(boolean b)       {this.reuse_addr=b; return this;}
 
 
     public void init() throws Exception {
@@ -123,11 +121,6 @@ public String getInfo() {
 
     public abstract void retainAll(Collection<Address> members);
 
-    /** BaseServer.Receiver interface */
-    public void receive(Address sender, byte[] data, int offset, int length) {
-        super.receive(sender, data, offset, length);
-    }
-
     public void receive(Address sender, ByteBuffer buf) {
         Util.bufferToArray(sender, buf, this);
     }

src/org/jgroups/protocols/TP.java

@@ -14,6 +14,7 @@
 import org.jgroups.util.ThreadFactory;
 import org.jgroups.util.UUID;
 
+import java.io.DataInput;
 import java.io.InterruptedIOException;
 import java.lang.management.ManagementFactory;
 import java.net.InetAddress;
@@ -1240,26 +1241,43 @@ public void receive(Address sender, byte[] data, int offset, int length) {
         if(Objects.equals(local_physical_addr, sender))
             return;
 
-        byte flags=data[Global.SHORT_SIZE];
-        boolean is_message_list=(flags & LIST) == LIST;
+        short version=Bits.readShort(data, offset);
+        if(!versionMatch(version, sender))
+            return;
+        offset+=Global.SHORT_SIZE;
+        byte flags=data[offset];
+        offset+=Global.BYTE_SIZE;
 
+        boolean is_message_list=(flags & LIST) == LIST, multicast=(flags & MULTICAST) == MULTICAST;
+        ByteArrayDataInputStream in=new ByteArrayDataInputStream(data, offset, length);
         if(is_message_list) // used if message bundling is enabled
-            handleMessageBatch(sender, data, offset, length);
+            handleMessageBatch(in, multicast);
         else
-            handleSingleMessage(sender, data, offset, length);
+            handleSingleMessage(in, multicast);
     }
 
+    public void receive(Address sender, DataInput in) throws Exception {
+        if(in == null) return;
 
-    protected void handleMessageBatch(Address sender, byte[] data, int offset, int length) {
-        try {
-            ByteArrayDataInputStream in=new ByteArrayDataInputStream(data, offset, length);
-            short version=in.readShort();
-            if(!versionMatch(version, sender))
-                return;
+        // drop message from self; it has already been looped back up (https://issues.jboss.org/browse/JGRP-1765)
+        if(Objects.equals(local_physical_addr, sender))
+            return;
+
+        short version=in.readShort();
+        if(!versionMatch(version, sender))
+            return;
+        byte flags=in.readByte();
+
+        boolean is_message_list=(flags & LIST) == LIST, multicast=(flags & MULTICAST) == MULTICAST;
+        if(is_message_list) // used if message bundling is enabled
+            handleMessageBatch(in, multicast);
+        else
+            handleSingleMessage(in, multicast);
+    }
 
-            byte flags=in.readByte();
-            final boolean multicast=(flags & MULTICAST) == MULTICAST;
 
+    protected void handleMessageBatch(DataInput in, boolean multicast) {
+        try {
             final MessageBatch[] batches=Util.readMessageBatch(in, multicast);
             final MessageBatch batch=batches[0], oob_batch=batches[1], internal_batch_oob=batches[2], internal_batch=batches[3];
 
@@ -1274,25 +1292,8 @@ protected void handleMessageBatch(Address sender, byte[] data, int offset, int l
     }
 
 
-    protected void processBatch(MessageBatch batch, boolean oob, boolean internal) {
+    protected void handleSingleMessage(DataInput in, boolean multicast) {
         try {
-            if(batch != null && !batch.isEmpty())
-                msg_processing_policy.process(batch, oob, internal);
-        }
-        catch(Throwable t) {
-            log.error("processing batch failed", t);
-        }
-    }
-
-    protected void handleSingleMessage(Address sender, byte[] data, int offset, int length) {
-        try {
-            ByteArrayDataInputStream in=new ByteArrayDataInputStream(data, offset, length);
-            short version=in.readShort();
-            if(!versionMatch(version, sender))
-                return;
-
-            byte flags=in.readByte();
-            final boolean multicast=(flags & MULTICAST) == MULTICAST;
             Message msg=new Message(false); // don't create headers, readFrom() will do this
             msg.readFrom(in);
 
@@ -1307,6 +1308,16 @@ protected void handleSingleMessage(Address sender, byte[] data, int offset, int
         }
     }
 
+    protected void processBatch(MessageBatch batch, boolean oob, boolean internal) {
+        try {
+            if(batch != null && !batch.isEmpty())
+                msg_processing_policy.process(batch, oob, internal);
+        }
+        catch(Throwable t) {
+            log.error("processing batch failed", t);
+        }
+    }
+
     public boolean unicastDestMismatch(Address dest) {
         return dest != null && !(Objects.equals(dest, local_addr) || Objects.equals(dest, local_physical_addr));
     }

src/org/jgroups/protocols/TUNNEL.java

@@ -179,9 +179,9 @@ public Object handleDownEvent(Event evt) {
     public void receive(GossipData data) {
         switch (data.getType()) {
             case MESSAGE:
-                byte[] msg=data.getBuffer();
                 if(Objects.equals(local_addr, data.getSender()))
                     return;
+                byte[] msg=data.getBuffer();
                 receive(data.getSender(), msg, 0, msg.length);
                 break;
             case SUSPECT:

src/org/jgroups/stack/GossipData.java

@@ -17,7 +17,7 @@
 
 
 /**
- * Encapsulates data sent between GossipRouter and GossipClient
+ * Encapsulates data sent between GossipRouter and RouterStub (TCPGOSSIP and TUNNEL)
  * @author Bela Ban Oct 4 2001
  */
 public class GossipData implements SizeStreamable {
@@ -88,12 +88,11 @@ public void setPingData(List<PingData> mbrs) {
         this.ping_data=mbrs;
     }
 
-    public GossipData addPingData(PingData data) {
+    public void addPingData(PingData data) {
         if(ping_data == null)
             ping_data=new ArrayList<>();
         if(data != null)
             ping_data.add(data);
-        return this;
     }
 
 
@@ -162,7 +161,13 @@ public void writeTo(DataOutput out) throws Exception {
     }
 
     public void readFrom(DataInput in) throws Exception {
-        type=GossipType.values()[in.readByte()];
+        readFrom(in, true);
+    }
+
+
+    protected void readFrom(DataInput in, boolean read_type) throws Exception {
+        if(read_type)
+            type=GossipType.values()[in.readByte()];
         group=Bits.readString(in);
         addr=Util.readAddress(in);
         sender=Util.readAddress(in);
@@ -189,4 +194,5 @@ public void readFrom(DataInput in) throws Exception {
     }
 
 
+
 }