- First batch of classes converted to use down(Message) [https://issu…

…es.jboss.org/browse/JGRP-2067]

- Protocols now use down_prot.down(msg) instead of down_prot.down(new Event(Event.MSG, msg))
- Converted up_prot.up(new Event(Event.MSG, msg) to up_prot.up(msg)
- Removed Event.MSG
- Catching exceptions when calling ThreadPoolExecutor.execute()

src/org/jgroups/Event.java

@@ -1,13 +1,13 @@
 package org.jgroups;
 
+
 /**
  * Event is a JGroups internal class used for inter-stack and intra-stack communication. 
  * 
  * @since 2.0
  * @author Bela Ban
  */
 public class Event {
-    public static final int MSG                                =  1;  // arg = Message
     public static final int CONNECT                            =  2;  // arg = cluster name (string)
     public static final int DISCONNECT                         =  4;  // arg = member address (Address)
     public static final int VIEW_CHANGE                        =  6;  // arg = View (or MergeView in case of merge)
@@ -97,7 +97,6 @@ public Event(int type, Object arg) {
 
     public static String type2String(int t) {
         switch(t) {
-            case MSG:	                 return "MSG";
             case CONNECT:	             return "CONNECT";
             case DISCONNECT:	         return "DISCONNECT";
             case VIEW_CHANGE:	         return "VIEW_CHANGE";
@@ -162,8 +161,6 @@ public static String type2String(int t) {
     public String toString() {
         StringBuilder ret=new StringBuilder(64);
         ret.append(type2String(type)).append(", arg=").append(arg);
-        if(type == MSG)
-            ret.append(" (headers=").append(((Message)arg).printHeaders()).append(")");
         return ret.toString();
     }
 

src/org/jgroups/JChannel.java

@@ -489,7 +489,7 @@ public JChannel send(Message msg) throws Exception {
         if(msg == null)
             throw new NullPointerException("msg is null");
         checkClosedOrNotConnected();
-        down(new Event(Event.MSG, msg));
+        down(msg);
         return this;
     }
 
@@ -641,11 +641,16 @@ public JChannel stopFlush(List<Address> flushParticipants) {
      */
     public Object down(Event evt) {
         if(evt == null) return null;
-        if(stats && evt.getType() == Event.MSG) {
+        return prot_stack.down(evt);
+    }
+
+    public Object down(Message msg) {
+        if(msg == null) return null;
+        if(stats) {
             sent_msgs++;
-            sent_bytes+=((Message)evt.getArg()).getLength();
+            sent_bytes+=msg.getLength();
         }
-        return prot_stack.down(evt);
+        return prot_stack.down(msg);
     }
 
 
@@ -656,19 +661,6 @@ public Object down(Event evt) {
      */
     public Object up(Event evt) {
         switch(evt.getType()) {
-
-            case Event.MSG:
-                Message msg=evt.getArg();
-                if(stats) {
-                    received_msgs++;
-                    received_bytes+=msg.getLength();
-                }
-
-                // discard local messages (sent by myself to me)
-                if(discard_own_messages && local_addr != null && msg.getSrc() != null && local_addr.equals(msg.getSrc()))
-                    return null;
-                break;
-
             case Event.VIEW_CHANGE:
                 View tmp=evt.getArg();
                 if(tmp instanceof MergeView)
@@ -776,6 +768,25 @@ public Object up(Event evt) {
         return null;
     }
 
+    public Object up(Message msg) {
+        if(stats) {
+            received_msgs++;
+            received_bytes+=msg.getLength();
+        }
+
+        // discard local messages (sent by myself to me)
+        if(discard_own_messages && local_addr != null && msg.getSrc() != null && local_addr.equals(msg.getSrc()))
+            return null;
+
+        // If UpHandler is installed, pass all events to it and return (UpHandler is e.g. a building block)
+        if(up_handler != null)
+            return up_handler.up(msg);
+
+        if(receiver != null)
+            receiver.receive(msg);
+        return null;
+    }
+
 
     /** Callback invoked by the protocol stack to deliver a message batch */
     public JChannel up(MessageBatch batch) {
@@ -908,9 +919,6 @@ protected JChannel getState(Address target, long timeout, Callable<Boolean> flus
 
     protected Object invokeCallback(int type, Object arg) {
         switch(type) {
-            case Event.MSG:
-                receiver.receive((Message)arg);
-                break;
             case Event.VIEW_CHANGE:
                 receiver.viewAccepted((View)arg);
                 break;

src/org/jgroups/UpHandler.java

@@ -4,30 +4,26 @@
 import org.jgroups.util.MessageBatch;
 
 /**
- * Provides a hook to hijack over all events received by a certain channel which has installed this
- * UpHandler.
- * <p>
- * Client usually never need to implement this interface and it is mostly used by JGroups building
- * blocks.
- * 
+ * Provides a hook to hijack all events received by a certain channel which has installed this UpHandler.<p>
+ * Client usually never need to implement this interface and it is mostly used by JGroups building blocks.
  * @since 2.0
  * @author Bela Ban
  */
 public interface UpHandler {
 
    /**
     * Invoked for all channel events except connection management and state transfer.
-    * @param evt
     */
     Object up(Event evt);
 
+    Object up(Message msg);
+
     default void up(MessageBatch batch) {
         for(Message msg: batch) {
             try {
-                up(new Event(Event.MSG, msg));
+                up(msg);
             }
             catch(Throwable t) {
-
             }
         }
     }

src/org/jgroups/auth/DemoToken.java

@@ -52,7 +52,7 @@ public boolean authenticate(AuthToken token, Message msg) {
         pending_requests.put(sender, entry); // here we'd have to check if a latch already exists...
         if(log.isTraceEnabled())
             log.trace(auth.getAddress() + ": sending challenge to " + sender);
-        auth.getDownProtocol().down(new Event(Event.MSG, challenge));
+        auth.getDownProtocol().down(challenge);
         try {
             entry.latch.await(block_time, TimeUnit.MILLISECONDS);
             pending_requests.remove(sender);
@@ -71,38 +71,33 @@ public void readFrom(DataInput in) throws Exception {}
     public int  size() {return 0;}
 
 
-    public boolean handleUpEvent(Event evt) {
-        switch(evt.getType()) {
-            case Event.MSG:
-                Message msg=(Message)evt.getArg();
-                DemoHeader hdr=(DemoHeader)msg.getHeader(ID);
-                if(hdr == null)
+    public boolean handleUpMessage(Message msg) {
+        DemoHeader hdr=msg.getHeader(ID);
+        if(hdr == null)
+            return true;
+        switch(hdr.type) {
+            case DemoHeader.CHALLENGE:
+                if(log.isTraceEnabled())
+                    log.trace(auth.getAddress() + ": received CHALLENGE from " + msg.getSrc());
+                long hash=hash(encrypt(hdr.payload));
+                Message response=new Message(msg.getSrc()).setFlag(Message.Flag.OOB);
+                response.putHeader(ID, new DemoHeader(hash));
+                if(log.isTraceEnabled())
+                    log.trace(auth.getAddress() + ": sending RESPONSE to " + msg.getSrc());
+                auth.getDownProtocol().down(response);
+                break;
+            case DemoHeader.RESPONSE:
+                if(log.isTraceEnabled())
+                    log.trace(auth.getAddress() + ": received RESPONSE from " + msg.getSrc());
+                Entry entry=pending_requests.get(msg.getSrc());
+                if(entry == null) {
+                    // error message
                     break;
-                switch(hdr.type) {
-                    case DemoHeader.CHALLENGE:
-                        if(log.isTraceEnabled())
-                            log.trace(auth.getAddress() + ": received CHALLENGE from " + msg.getSrc());
-                        long hash=hash(encrypt(hdr.payload));
-                        Message response=new Message(msg.getSrc()).setFlag(Message.Flag.OOB);
-                        response.putHeader(ID, new DemoHeader(hash));
-                        if(log.isTraceEnabled())
-                            log.trace(auth.getAddress() + ": sending RESPONSE to " + msg.getSrc());
-                        auth.getDownProtocol().down(new Event(Event.MSG, response));
-                        break;
-                    case DemoHeader.RESPONSE:
-                        if(log.isTraceEnabled())
-                            log.trace(auth.getAddress() + ": received RESPONSE from " + msg.getSrc());
-                        Entry entry=pending_requests.get(msg.getSrc());
-                        if(entry == null) {
-                            // error message
-                            break;
-                        }
-                        entry.setResponse(hdr.hash);
-                        break;
                 }
-                return false; // don't pass up
+                entry.setResponse(hdr.hash);
+                break;
         }
-        return true;
+        return false; // don't pass up
     }
 
 

src/org/jgroups/blocks/MessageDispatcher.java

@@ -574,6 +574,12 @@ public Object up(Event evt) {
             return null;
         }
 
+        public Object up(Message msg) {
+            if(corr != null)
+                corr.receiveMessage(msg);
+            return null;
+        }
+
         public void up(MessageBatch batch) {
             if(corr == null)
                 return;
@@ -582,17 +588,20 @@ public void up(MessageBatch batch) {
 
         @Override
         public Object down(Event evt) {
+            return channel != null? channel.down(evt) : null;
+        }
+
+        public Object down(Message msg) {
             if(channel != null) {
-                if(evt.getType() == Event.MSG && !(channel.isConnected() || channel.isConnecting())) {
+                if(!(channel.isConnected() || channel.isConnecting())) {
                     // return null;
                     throw new IllegalStateException("channel is not connected");
                 }
-                return channel.down(evt);
+                return channel.down(msg);
             }
             return null;
         }
 
-
         /* ----------------------- End of Protocol Interface ------------------------ */
 
     }

src/org/jgroups/blocks/RequestCorrelator.java

@@ -156,7 +156,7 @@ public void sendRequest(Collection<Address> dest_mbrs, Buffer data, Request req,
 
         if(opts.anycasting()) {
             if(opts.useAnycastAddresses()) {
-                transport.down(new Event(Event.MSG, msg.dest(new AnycastAddress(dest_mbrs))));
+                transport.down(msg.dest(new AnycastAddress(dest_mbrs)));
             }
             else {
                 boolean first=true;
@@ -165,12 +165,12 @@ public void sendRequest(Collection<Address> dest_mbrs, Buffer data, Request req,
                     first=false;
                     if(!mbr.equals(local_addr) && copy.isTransientFlagSet(Message.TransientFlag.DONT_LOOPBACK))
                         copy.clearTransientFlag(Message.TransientFlag.DONT_LOOPBACK);
-                    transport.down(new Event(Event.MSG, copy));
+                    transport.down(copy);
                 }
             }
         }
         else
-            transport.down(new Event(Event.MSG, msg));
+            transport.down(msg);
     }
 
     /** Sends a request to a single destination */
@@ -198,7 +198,7 @@ public void sendUnicastRequest(Address dest, Buffer data, Request req, RequestOp
         }
         else // async RPC
             rpc_stats.add(RpcStats.Type.UNICAST, dest, false, 0);
-        transport.down(new Event(Event.MSG, msg));
+        transport.down(msg);
     }
 
 
@@ -234,10 +234,6 @@ public boolean receive(Event evt) {
                 setLocalAddress(evt.getArg());
                 break;
 
-            case Event.MSG:
-                if(receiveMessage(evt.getArg()))
-                    return true; // message was consumed, don't pass it up
-                break;
             case Event.SITE_UNREACHABLE:
                 SiteMaster site_master=evt.getArg();
                 String site=site_master.getSite();
@@ -452,7 +448,7 @@ protected void sendResponse(Message rsp, long req_id, boolean is_exception) {
         rsp.putHeader(corr_id, rsp_hdr);
         if(log.isTraceEnabled())
             log.trace("sending rsp for %d to %s", req_id, rsp.getDest());
-        transport.down(new Event(Event.MSG, rsp));
+        transport.down(rsp);
     }
 
     protected static Buffer replyToBuffer(Object obj, Marshaller marshaller) throws Exception {

src/org/jgroups/fork/ForkChannel.java

@@ -197,17 +197,16 @@ public void close() {
         ((ForkProtocolStack)prot_stack).remove(fork_channel_id);
         if(state == State.CLOSED)
             return;
-        disconnect();                     // leave group if connected
+        disconnect();  // leave cluster if connected
         prot_stack.destroy();
         state=State.CLOSED;
         notifyChannelClosed(this);
     }
 
     @Override
-    public Object down(Event evt) {
-        if(evt.getType() == Event.MSG)
-            setHeader(evt.getArg());
-        return super.down(evt);
+    public Object down(Message msg) {
+        setHeader(msg);
+        return super.down(msg);
     }
 
     @Override
@@ -220,7 +219,7 @@ public ForkChannel send(Message msg) throws Exception {
             hdr=new FORK.ForkHeader(null, fork_channel_id);
             msg.putHeader(FORK.ID, hdr);
         }
-        prot_stack.down(new Event(Event.MSG, msg));
+        prot_stack.down(msg);
         return this;
     }
 

src/org/jgroups/fork/ForkProtocol.java

@@ -27,14 +27,6 @@ public Map<String,Object> dumpStats() {
 
     public Object down(Event evt) {
         switch(evt.getType()) {
-            case Event.MSG:
-                Message msg=(Message)evt.getArg();
-                FORK.ForkHeader hdr=(FORK.ForkHeader)msg.getHeader(FORK.ID);
-                if(hdr == null)
-                    msg.putHeader(FORK.ID, hdr=new FORK.ForkHeader(fork_stack_id, null));
-                else
-                    hdr.setForkStackId(fork_stack_id);
-                break;
             case Event.SET_LOCAL_ADDRESS:
             case Event.VIEW_CHANGE:
             case Event.CONNECT:
@@ -46,4 +38,13 @@ public Object down(Event evt) {
         }
         return down_prot.down(evt);
     }
+
+    public Object down(Message msg) {
+        FORK.ForkHeader hdr=msg.getHeader(FORK.ID);
+        if(hdr == null)
+            msg.putHeader(FORK.ID, hdr=new FORK.ForkHeader(fork_stack_id, null));
+        else
+            hdr.setForkStackId(fork_stack_id);
+        return down_prot.down(msg);
+    }
 }