CVS update: JGroups/src/org/jgroups/client StompConnection.java

  User: belaban 
  Date: 10/10/26 12:30:16

  Modified:    src/org/jgroups/client StompConnection.java
  Log:
  ns
  
  Revision  Changes    Path
  1.2       +115 -9    JGroups/src/org/jgroups/client/StompConnection.java
  
  Index: StompConnection.java
  ===================================================================
  RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/client/StompConnection.java,v
  retrieving revision 1.1
  retrieving revision 1.2
  diff -u -r1.1 -r1.2
  --- StompConnection.java	26 Oct 2010 06:46:51 -0000	1.1
  +++ StompConnection.java	26 Oct 2010 12:30:16 -0000	1.2
  @@ -1,5 +1,8 @@
   package org.jgroups.client;
   
  +import org.jgroups.logging.Log;
  +import org.jgroups.logging.LogFactory;
  +import org.jgroups.protocols.STOMP;
   import org.jgroups.util.Util;
   
   import java.io.DataInputStream;
  @@ -13,9 +16,9 @@
   /**
    * STOMP client to access the STOMP protocol
    * @author Bela Ban
  - * @version $Id: StompConnection.java,v 1.1 2010/10/26 06:46:51 belaban Exp $
  + * @version $Id: StompConnection.java,v 1.2 2010/10/26 12:30:16 belaban Exp $
    */
  -public class StompConnection {
  +public class StompConnection implements Runnable {
       protected Socket           sock;
       protected DataInputStream  in;
       protected DataOutputStream out;
  @@ -25,6 +28,13 @@
   
       protected final Set<Listener> listeners=new HashSet<Listener>();
   
  +    protected final Set<String> subscriptions=new HashSet<String>();
  +
  +    protected Thread runner;
  +
  +    protected final Log log=LogFactory.getLog(getClass());
  +
  +
       /**
        *
        * @param dest IP address + ':' + port, e.g. "192.168.1.5:8787"
  @@ -47,17 +57,38 @@
       public void connect(String userid, String password) throws IOException {
           String dest;
   
  +        if(isConnected())
  +            return;
           while((dest=pickRandomDestination()) != null) {
               try {
                   connect(dest);
  -                return;
  +                break;
               }
               catch(IOException ex) {
                   close();
                   server_destinations.remove(dest);
               }
           }
  -        throw new IOException("no target server available");
  +        if(!isConnected())
  +            throw new IOException("no target server available");
  +
  +        StringBuilder sb=new StringBuilder();
  +        sb.append(STOMP.ClientVerb.CONNECT.name()).append("\n");
  +        if(userid != null)
  +            sb.append("login: ").append(userid).append("\n");
  +        if(password != null)
  +            sb.append("passcode: ").append(password).append("\n");
  +        sb.append("\n").append(STOMP.NULL_BYTE);
  +
  +        out.write(sb.toString().getBytes());
  +        out.flush();
  +    }
  +
  +
  +    public void reconnect() throws IOException {
  +        connect();
  +        for(String subscription: subscriptions)
  +            subscribe(subscription);
       }
   
   
  @@ -71,11 +102,15 @@
       }
   
       public void subscribe(String destination) {
  -
  +        if(destination == null)
  +            return;
  +        subscriptions.add(destination);
       }
   
       public void unsubscribe(String destination) {
  -
  +        if(destination == null)
  +            return;
  +        subscriptions.remove(destination);
       }
   
       public void send(String destination, byte[] buf, int offset, int length) {
  @@ -86,7 +121,57 @@
           send(destination, buf, 0, buf.length);
       }
   
  -    
  +    public void run() {
  +        while(isConnected()) {
  +            try {
  +                STOMP.Frame frame=STOMP.readFrame(in);
  +                if(frame != null) {
  +                    STOMP.ServerVerb verb=STOMP.ServerVerb.valueOf(frame.getVerb());
  +                    System.out.println("frame = " + frame);
  +                    switch(verb) {
  +                        case MESSAGE:
  +                            byte[] buf=frame.getBody();
  +                            notifyListeners(frame.getHeaders(), buf, 0, buf != null? buf.length : 0);
  +                            break;
  +                        case CONNECTED:
  +                            break;
  +                        case ERROR:
  +                            break;
  +                        case INFO:
  +                            break;
  +                        case RECEIPT:
  +                            break;
  +                        default:
  +                            throw new IllegalArgumentException("verb " + verb + " is not known");
  +                    }
  +                }
  +            }
  +            catch(IOException e) {
  +                close();
  +                try {
  +                    reconnect();
  +                }
  +                catch(IOException e1) {
  +                    log.warn("failed to reconnect; runner thread terminated");
  +                }
  +            }
  +            catch(Throwable t) {
  +                log.error("failure reading frame", t);
  +            }
  +        }
  +    }
  +
  +    protected void notifyListeners(Map<String,String> headers, byte[] buf, int offset, int length) {
  +        for(Listener listener: listeners) {
  +            try {
  +                listener.onMessage(headers, buf, offset, length);
  +            }
  +            catch(Throwable t) {
  +                log.error("failed calling listener", t);
  +            }
  +        }
  +    }
  +
       protected String pickRandomDestination() {
           return server_destinations.isEmpty()? null : server_destinations.iterator().next();
       }
  @@ -97,6 +182,7 @@
           sock.connect(saddr);
           in=new DataInputStream(sock.getInputStream());
           out=new DataOutputStream(sock.getOutputStream());
  +        startRunner();
       }
   
       protected static SocketAddress parse(String dest) throws UnknownHostException {
  @@ -116,9 +202,17 @@
           return sock != null && sock.isConnected();
       }
   
  +    protected synchronized void startRunner() {
  +        if(runner == null || !runner.isAlive()) {
  +            runner=new Thread(this, "StompConnection receiver");
  +            runner.start();
  +        }
  +    }
  +
  +
   
       public static interface Listener {
  -        void onMessage(byte[] buf, int offset, int length);
  +        void onMessage(Map<String,String> headers, byte[] buf, int offset, int length);
           void onInfo(Map<String,String> information);
       }
   
  @@ -140,7 +234,19 @@
               return;
           }
           StompConnection conn=new StompConnection(host+ ":" + port);
  +        conn.addListener(new Listener() {
  +
  +            public void onMessage(Map<String, String> headers, byte[] buf, int offset, int length) {
  +                System.out.println("<< " + new String(buf, offset, length) + ", headers: " + headers);
  +            }
  +
  +            public void onInfo(Map<String, String> information) {
  +            }
  +        });
           conn.connect();
  -        
  +
  +        for(;;) {
  +            
  +        }
       }
   }
  
  
  

------------------------------------------------------------------------------
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