CVS update: JGroups/src/org/jgroups/protocols STOMP.java

  User: belaban 
  Date: 10/10/22 13:23:52

  Modified:    src/org/jgroups/protocols STOMP.java
  Log:
  ns
  
  Revision  Changes    Path
  1.6       +34 -7     JGroups/src/org/jgroups/protocols/STOMP.java
  
  Index: STOMP.java
  ===================================================================
  RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/protocols/STOMP.java,v
  retrieving revision 1.5
  retrieving revision 1.6
  diff -u -r1.5 -r1.6
  --- STOMP.java	21 Oct 2010 14:16:25 -0000	1.5
  +++ STOMP.java	22 Oct 2010 13:23:52 -0000	1.6
  @@ -13,16 +13,14 @@
   import java.net.ServerSocket;
   import java.net.Socket;
   import java.nio.ByteBuffer;
  -import java.util.HashMap;
  -import java.util.LinkedList;
  -import java.util.List;
  -import java.util.Map;
  +import java.util.*;
  +import java.util.concurrent.ConcurrentMap;
   
   /**
    * Protocol which provides STOMP support. Very simple implementation, with a 1 thread / connection model. Use for
    * a few hundred clients max.
    * @author Bela Ban
  - * @version $Id: STOMP.java,v 1.5 2010/10/21 14:16:25 belaban Exp $
  + * @version $Id: STOMP.java,v 1.6 2010/10/22 13:23:52 belaban Exp $
    * @since 2.11
    */
   @MBean
  @@ -36,7 +34,13 @@
   
       /* ---------------------------------------------   JMX      ---------------------------------------------------*/
       @ManagedAttribute(description="Number of client connections",writable=false)
  -    int getNumConnections() {return connections.size();}
  +    public int getNumConnections() {return connections.size();}
  +
  +    @ManagedAttribute(description="Number of subscriptions",writable=false)
  +    public int getNumSubscriptions() {return subscriptions.size();}
  +
  +    @ManagedAttribute(description="Print subscriptions",writable=false)
  +    public String getSubscriptions() {return subscriptions.keySet().toString();}
   
   
       /* --------------------------------------------- Fields ------------------------------------------------------ */
  @@ -44,6 +48,9 @@
       protected Thread                 acceptor;
       protected final List<Connection> connections=new LinkedList<Connection>();
   
  +    // Subscriptions and connections which are subscribed
  +    protected final ConcurrentMap<String, Set<Connection>> subscriptions=Util.createConcurrentMap(20);
  +
   
   
       public static enum ClientVerb      {CONNECT, SUBSCRIBE, UNSUBSCRIBE, BEGIN, COMMIT, ABORT, ACK, DISCONNECT};
  @@ -167,7 +174,6 @@
                   catch(Throwable t) {
                       log.error("failure reading frame", t);
                   }
  -
               }
           }
   
  @@ -180,8 +186,29 @@
                                     "password-check", "none");
                       break;
                   case SUBSCRIBE:
  +                    Map<String,String> headers=frame.getHeaders();
  +                    String destination=headers.get("destination");
  +                    if(destination != null) {
  +                        Set<Connection> conns=subscriptions.get(destination);
  +                        if(conns == null) {
  +                            conns=new HashSet<Connection>();
  +                            Set<Connection> tmp=subscriptions.putIfAbsent(destination, conns);
  +                            if(tmp != null)
  +                                conns=tmp;
  +                        }
  +                        conns.add(this);
  +                    }
                       break;
                   case UNSUBSCRIBE:
  +                    headers=frame.getHeaders();
  +                    destination=headers.get("destination");
  +                    if(destination != null) {
  +                        Set<Connection> conns=subscriptions.get(destination);
  +                        if(conns != null) {
  +                            if(conns.remove(this) && conns.isEmpty())
  +                                subscriptions.remove(destination);
  +                        }
  +                    }
                       break;
                   case BEGIN:
                       break;
  
  
  

------------------------------------------------------------------------------
Nokia and AT&T present the 2010 Calling All Innovators-North America contest
Create new apps & games for the Nokia N8 for consumers in  U.S. and Canada
$10 million total in prizes - $4M cash, 500 devices, nearly $6M in marketing
Develop with Nokia Qt SDK, Web Runtime, or Java and Publish to Ovi Store 
http://p.sf.net/sfu/nokia-dev2dev