CVS update: JGroups/src/org/jgroups/protocols STOMP.java
"Bela Ban" <[email protected]> Thu, 21 Oct 2010 13:12:26 +0000
| Newsgroups | gmane.comp.java.javagroups.cvs |
|---|---|
| Message-ID | <[email protected]> |
User: belaban
Date: 10/10/21 13:12:26
Modified: src/org/jgroups/protocols STOMP.java
Log:
added frame parsing
Revision Changes Path
1.2 +112 -8 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.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- STOMP.java 21 Oct 2010 10:45:53 -0000 1.1
+++ STOMP.java 21 Oct 2010 13:12:25 -0000 1.2
@@ -1,37 +1,42 @@
package org.jgroups.protocols;
import org.jgroups.Global;
-import org.jgroups.annotations.Experimental;
-import org.jgroups.annotations.MBean;
-import org.jgroups.annotations.Property;
-import org.jgroups.annotations.Unsupported;
+import org.jgroups.annotations.*;
import org.jgroups.stack.Protocol;
import org.jgroups.util.Util;
import java.io.DataInputStream;
import java.io.DataOutputStream;
+import java.io.EOFException;
import java.io.IOException;
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;
/**
* 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.1 2010/10/21 10:45:53 belaban Exp $
+ * @version $Id: STOMP.java,v 1.2 2010/10/21 13:12:25 belaban Exp $
* @since 2.11
*/
@MBean
@Experimental @Unsupported
public class STOMP extends Protocol implements Runnable {
- /* ----------------------------------------- Properties -------------------------------------------------- */
+ /* ----------------------------------------- Properties ----------------------------------------------- */
@Property(description="Port on which the STOMP protocol listens for requests",writable=false)
protected int port=8787;
+ /* --------------------------------------------- JMX ---------------------------------------------------*/
+ @ManagedAttribute(description="Number of client connections",writable=false)
+ int getNumConnections() {return connections.size();}
+
/* --------------------------------------------- Fields ------------------------------------------------------ */
protected ServerSocket srv_sock;
@@ -83,8 +88,7 @@
super.stop();
}
- // acceptor loop
-
+ // Acceptor loop
public void run() {
Socket client_sock;
while(acceptor != null && srv_sock != null) {
@@ -130,10 +134,110 @@
Util.close(in);
Util.close(out);
Util.close(sock);
+ synchronized(connections) {
+ connections.remove(this);
+ }
}
public void run() {
+ while(!sock.isClosed()) {
+ try {
+ Frame frame=readFrame(in);
+ System.out.println("frame = " + frame);
+ }
+ catch(IOException ex) {
+ log.error("failure reading frame", ex);
+ stop(); // ??
+ }
+
+ }
+ }
+
+ private Frame readFrame(DataInputStream in) throws IOException {
+ String verb=Util.readLine(in);
+ if(verb == null)
+ throw new EOFException("reading verb");
+
+ Map<String,String> headers=new HashMap<String,String>();
+ byte[] body=null;
+
+ for(;;) {
+ String header=Util.readLine(in);
+ if(header == null)
+ throw new EOFException("reading header");
+ if(header.length() == 0)
+ break;
+ int index=header.indexOf(":");
+ if(index != -1)
+ headers.put(header.substring(0, index).trim(), header.substring(index+1).trim());
+ }
+
+ if(headers.containsKey("length")) {
+ int length=Integer.parseInt(headers.get("length"));
+ body=new byte[length];
+ in.read(body, 0, body.length);
+ }
+ else {
+ ByteBuffer buf=ByteBuffer.allocate(500);
+ boolean terminate=false;
+ for(;;) {
+ int c=in.read();
+ if(c == -1 || c == 0)
+ terminate=true;
+
+ if(buf.remaining() == 0 || terminate) {
+ if(body == null) {
+ body=new byte[buf.position()];
+ System.arraycopy(buf.array(), buf.arrayOffset(), body, 0, buf.position());
+ }
+ else {
+ byte[] tmp=new byte[body.length + buf.position()];
+ System.arraycopy(body, 0, tmp, 0, body.length);
+ try {
+ System.arraycopy(buf.array(), buf.arrayOffset(), tmp, body.length, buf.position());
+ }
+ catch(Throwable t) {
+ }
+ body=tmp;
+ }
+ buf.rewind();
+ }
+
+ if(terminate)
+ break;
+
+ buf.put((byte)c);
+ }
+ }
+
+
+ return new Frame(verb, headers, body);
+ }
+ }
+
+ protected static class Frame {
+ final String verb;
+ final Map<String,String> headers;
+ final byte[] body;
+
+ public Frame(String verb, Map<String, String> headers, byte[] body) {
+ this.verb=verb;
+ this.headers=headers;
+ this.body=body;
+ }
+
+ public String toString() {
+ StringBuilder sb=new StringBuilder();
+ sb.append(verb).append("\n");
+ if(headers != null && !headers.isEmpty()) {
+ for(Map.Entry<String,String> entry: headers.entrySet())
+ sb.append(entry.getKey()).append(": ").append(entry.getValue()).append("\n");
+ }
+ if(body != null && body.length > 0) {
+ sb.append("body: ").append(body.length).append(" bytes");
+ }
+ return sb.toString();
}
}
}
------------------------------------------------------------------------------
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