- Created ReliableUnicast, UNICAST4 (https://issues.redhat.com/browse…

…/JGRP-2843)

- Added add(List<Tuple>...) to Buffer and implementations, plus tests
- Added UNICAST4 to UnicastUnitTest
- UNICAST3_Test -> UNICAST_Test (also tests UNICAST4)
- Added UNICAST4 to UNICAST_ConnectionTests
- Added UNICAST4 to UNICAST_ContentionTest
- Added UNICAST4 to UNICAST_OOB_Test
- Added UNICAST4 to UNICAST_MessagesToSelfTest
- Added UNICAST4 to UNICAST_DropFirstAndLastTest
- Added REVERSE2
- Simplified MessageCache (used by UNICAST-X protocols)
- Fixed serialization of Average
- Fixed getReceiverEntry() (https://issues.redhat.com/browse/JGRP-2852)
- Added new MockTransport
- Added tests to ReliableUnicastTest (https://issues.redhat.com/browse/JGRP-2852)
- Changed ReliableUnicast.up(MessageBatch): conn-ids are handled after removing unicast messages from the batch into a hashmap (https://issues.redhat.com/browse/JGRP-2852)

conf/jg-magic-map.xml

@@ -62,6 +62,7 @@
     <class id="97"  name="org.jgroups.protocols.RTTHeader"/>
     <class id="98"  name="org.jgroups.protocols.ProtPerfHeader"/>
     <class id="99"  name="org.jgroups.protocols.NakAckHeader"/>
+    <class id="100" name="org.jgroups.protocols.UnicastHeader"/>
 
 </magic-number-class-mapping>
 

conf/jg-protocol-ids.xml

@@ -64,6 +64,7 @@
     <class id="75" name="org.jgroups.protocols.JDBC_PING2"/>
     <class id="76" name="org.jgroups.protocols.NAKACK3"/>
     <class id="77" name="org.jgroups.protocols.NAKACK4"/>
+    <class id="78" name="org.jgroups.protocols.UNICAST4"/>
 
     <!-- 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/ObjectMessage.java

@@ -73,7 +73,7 @@ public ObjectMessage(Address dest, SizeStreamable obj) {
     public ObjectMessage     setArray(ByteArray buf)              {throw new UnsupportedOperationException();}
     public boolean           isWrapped()                          {return isFlagSet(Flag.SERIALIZED);}
 
-    // reusing SERIALIZABLE
+    // reusing SERIALIZED
     public ObjectMessage setWrapped(boolean b) {
         if(b) setFlag(Flag.SERIALIZED);
         else  clearFlag(Flag.SERIALIZED);
@@ -145,8 +145,11 @@ public void readPayload(DataInput in) throws IOException, ClassNotFoundException
     }
 
     @Override protected Message copyPayload(Message copy) {
-        if(obj != null)
+        if(obj != null) {
             ((ObjectMessage)copy).setObject(obj);
+            if(isFlagSet(Flag.SERIALIZED))
+                copy.setFlag(Flag.SERIALIZED);
+        }
         return copy;
     }
 

src/org/jgroups/conf/ClassConfigurator.java

@@ -30,7 +30,7 @@ public class ClassConfigurator {
     protected static final String ID                = "id";
     protected static final String NAME              = "name";
     protected static final String EXTERNAL          = "external";
-    private static final int      MAX_MAGIC_VALUE=100;
+    private static final int      MAX_MAGIC_VALUE=124;
     private static final int      MAX_PROT_ID_VALUE=256;
     private static final short    MIN_CUSTOM_MAGIC_NUMBER=1024;
     private static final short    MIN_CUSTOM_PROTOCOL_ID=512;
@@ -292,7 +292,7 @@ protected static void alreadyInProtocolsMap(short prot_id, String classname) {
      * try to read the magic number configuration file as a Resource form the classpath using getResourceAsStream
      * if this fails this method tries to read the configuration file from mMagicNumberFile using a FileInputStream (not in classpath but somewhere else in the disk)
      *
-     * @return an array of ClassMap objects that where parsed from the file (if found) or an empty array if file not found or had en exception
+     * @return a list of ClassMap objects that where parsed from the file (if found) or an empty array if file not found or had en exception
      */
     protected static List<Triple<Short,String,Boolean>> readMappings(String name) throws Exception {
         InputStream stream=Util.getResourceAsStream(name, ClassConfigurator.class);

src/org/jgroups/protocols/NAKACK4.java

@@ -19,7 +19,7 @@
 import static org.jgroups.conf.AttributeType.SCALAR;
 
 /**
- * New multicast protocols based on fixed-size xmit windows and message ACKs<rb/>
+ * New multicast protocol based on fixed-size xmit windows and message ACKs<br/>
  * Details: https://issues.redhat.com/browse/JGRP-2780
  * @author Bela Ban
  * @since  5.4

src/org/jgroups/protocols/REVERSE2.java

@@ -0,0 +1,66 @@
+package org.jgroups.protocols;
+
+import org.jgroups.Message;
+import org.jgroups.annotations.MBean;
+import org.jgroups.stack.Protocol;
+import org.jgroups.util.MessageBatch;
+
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Predicate;
+
+/**
+ * @author Bela Ban
+ * @since x.y
+ */
+@MBean(description="Reverts messages based on a filter and delivers them when told")
+public class REVERSE2 extends Protocol {
+    protected volatile Predicate<Message> filter; // if set and true: queue messages
+    protected final Deque<Message>        queue=new ConcurrentLinkedDeque<>();
+
+    public Predicate<Message> filter()                     {return filter;}
+    public REVERSE2           filter(Predicate<Message> f) {this.filter=f; return this;}
+    public int                size()                       {return queue.size();}
+
+    /** Delivers queued messages */
+    public int deliver() {
+        Message msg; int count=0;
+        while((msg=queue.pollLast()) != null) {
+            up_prot.up(msg);
+            count++;
+        }
+        return count;
+    }
+
+    @Override
+    public Object up(Message msg) {
+        if(filter != null && filter.test(msg)) {
+            queue.add(msg);
+            return null;
+        }
+        return up_prot.up(msg);
+    }
+
+    @Override
+    public void up(MessageBatch batch) {
+        if(filter == null) {
+            up_prot.up(batch);
+            return;
+        }
+        for(Iterator<Message> it=batch.iterator(); it.hasNext();) {
+            Message msg=it.next();
+            if(filter.test(msg)) {
+                it.remove();
+                queue.add(msg);
+            }
+        }
+        if(!batch.isEmpty())
+            up_prot.up(batch);
+    }
+
+    @Override
+    public String toString() {
+        return String.format("%d msgs", queue.size());
+    }
+}