Skip to content

Commit 189aad9

Browse files
author
Kai Hackemesser
committed
Added support for RabbitMQ Stomp adapter, feature Durable Topic Subscription
1 parent 4152479 commit 189aad9

2 files changed

Lines changed: 96 additions & 16 deletions

File tree

‎stompjms-client/src/main/java/org/fusesource/stomp/jms/RabbitMQServerAdaptor.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package org.fusesource.stomp.jms;
2+
3+
import static org.fusesource.stomp.client.Constants.ID;
4+
import static org.fusesource.stomp.client.Constants.PERSISTENT;
5+
import static org.fusesource.stomp.client.Constants.TRUE;
6+
import static org.fusesource.stomp.client.Constants.UNSUBSCRIBE;
7+
8+
import java.util.Map;
9+
10+
import javax.jms.JMSException;
11+
12+
import org.fusesource.hawtbuf.AsciiBuffer;
13+
import org.fusesource.stomp.codec.StompFrame;
14+
15+
/**
16+
* Stomp adaptor for RabbitMQ
17+
*/
18+
public class RabbitMQServerAdaptor extends StompServerAdaptor {
19+
20+
@Override
21+
public boolean matchesServerAndVersion(String server) {
22+
return server.startsWith("RabbitMQ/");
23+
}
24+
25+
@Override
26+
public void addSubscribeHeaders(Map<AsciiBuffer, AsciiBuffer> headerMap, boolean persistent, boolean browser,
27+
boolean noLocal, StompJmsPrefetch prefetch) throws JMSException {
28+
if (browser) {
29+
throw new JMSException("Server does not support browsing over STOMP");
30+
}
31+
if (noLocal) {
32+
throw new JMSException("Server does not support 'no local' semantics over STOMP");
33+
}
34+
if (persistent) {
35+
headerMap.put(PERSISTENT, TRUE);
36+
}
37+
}
38+
39+
@Override
40+
public StompFrame createUnsubscribeFrame(AsciiBuffer consumerId, boolean persistent) throws JMSException {
41+
StompFrame frame = new StompFrame();
42+
Map<AsciiBuffer, AsciiBuffer> headerMap = frame.headerMap();
43+
frame.action(UNSUBSCRIBE);
44+
headerMap.put(ID, consumerId);
45+
if (persistent) {
46+
headerMap.put(PERSISTENT, TRUE);
47+
}
48+
return frame;
49+
}
50+
}

‎stompjms-client/src/main/java/org/fusesource/stomp/jms/StompChannel.java‎

Lines changed: 46 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,20 +10,27 @@
1010

1111
package org.fusesource.stomp.jms;
1212

13-
import org.fusesource.hawtbuf.AsciiBuffer;
14-
import org.fusesource.hawtdispatch.Task;
15-
import org.fusesource.stomp.client.CallbackConnection;
16-
import org.fusesource.stomp.client.ProtocolException;
17-
import org.fusesource.stomp.client.Stomp;
18-
import org.fusesource.stomp.codec.StompFrame;
19-
import org.fusesource.stomp.client.Callback;
20-
import org.fusesource.stomp.client.Promise;
21-
import org.fusesource.stomp.jms.message.StompJmsMessage;
22-
import org.fusesource.stomp.jms.util.StompTranslator;
13+
import static org.fusesource.hawtdispatch.Dispatch.NOOP;
14+
import static org.fusesource.stomp.client.Constants.ABORT;
15+
import static org.fusesource.stomp.client.Constants.ACK;
16+
import static org.fusesource.stomp.client.Constants.ACK_MODE;
17+
import static org.fusesource.stomp.client.Constants.BEGIN;
18+
import static org.fusesource.stomp.client.Constants.COMMIT;
19+
import static org.fusesource.stomp.client.Constants.CONTENT_LENGTH;
20+
import static org.fusesource.stomp.client.Constants.DESTINATION;
21+
import static org.fusesource.stomp.client.Constants.DISCONNECT;
22+
import static org.fusesource.stomp.client.Constants.HOST_ID;
23+
import static org.fusesource.stomp.client.Constants.ID;
24+
import static org.fusesource.stomp.client.Constants.MESSAGE;
25+
import static org.fusesource.stomp.client.Constants.MESSAGE_ID;
26+
import static org.fusesource.stomp.client.Constants.SELECTOR;
27+
import static org.fusesource.stomp.client.Constants.SEND;
28+
import static org.fusesource.stomp.client.Constants.SERVER;
29+
import static org.fusesource.stomp.client.Constants.SESSION;
30+
import static org.fusesource.stomp.client.Constants.SUBSCRIBE;
31+
import static org.fusesource.stomp.client.Constants.SUBSCRIPTION;
32+
import static org.fusesource.stomp.client.Constants.TRANSACTION;
2333

24-
import javax.jms.ExceptionListener;
25-
import javax.jms.JMSException;
26-
import javax.net.ssl.SSLContext;
2734
import java.io.IOException;
2835
import java.net.URI;
2936
import java.util.Map;
@@ -33,14 +40,27 @@
3340
import java.util.concurrent.atomic.AtomicBoolean;
3441
import java.util.concurrent.atomic.AtomicInteger;
3542

36-
import static org.fusesource.stomp.client.Constants.*;
37-
import static org.fusesource.hawtdispatch.Dispatch.*;
43+
import javax.jms.ExceptionListener;
44+
import javax.jms.JMSException;
45+
import javax.net.ssl.SSLContext;
46+
47+
import org.fusesource.hawtbuf.AsciiBuffer;
48+
import org.fusesource.hawtdispatch.Task;
49+
import org.fusesource.stomp.client.Callback;
50+
import org.fusesource.stomp.client.CallbackConnection;
51+
import org.fusesource.stomp.client.Promise;
52+
import org.fusesource.stomp.client.ProtocolException;
53+
import org.fusesource.stomp.client.Stomp;
54+
import org.fusesource.stomp.codec.StompFrame;
55+
import org.fusesource.stomp.jms.message.StompJmsMessage;
56+
import org.fusesource.stomp.jms.util.StompTranslator;
3857

3958
public class StompChannel {
4059

4160
private static final StompServerAdaptor STOMP_SERVER_ADAPTORS[] = new StompServerAdaptor[]{
4261
new ApolloServerAdaptor(),
4362
new ActiveMQServerAdaptor(),
63+
new RabbitMQServerAdaptor(),
4464
new StompServerAdaptor()
4565
};
4666

@@ -105,11 +125,14 @@ public void connect() throws JMSException {
105125
connection = future.await();
106126
writeBufferRemaining.set(connection.transport().getProtocolCodec().getWriteBufferSize());
107127
connection.getDispatchQueue().execute(new Task() {
128+
@Override
108129
public void run() {
109130
connection.receive(new Callback<StompFrame>() {
131+
@Override
110132
public void onFailure(Throwable value) {
111133
handleException(value);
112134
}
135+
@Override
113136
public void onSuccess(StompFrame value) {
114137
onFrame(value);
115138
}
@@ -158,12 +181,15 @@ public void close() throws JMSException {
158181

159182
// Request a DISCONNECT so that we can try to flush the socket out.
160183
connection.getDispatchQueue().execute(new Task(){
184+
@Override
161185
public void run() {
162186
StompFrame frame = new StompFrame(DISCONNECT);
163187
connection.request(frame, new Callback<StompFrame>(){
188+
@Override
164189
public void onFailure(Throwable value) {
165190
onSuccess(null);
166191
}
192+
@Override
167193
public void onSuccess(StompFrame value) {
168194
cd.countDown();
169195
}
@@ -321,8 +347,10 @@ public void sendFrame(final StompFrame frame) throws IOException {
321347
if( writeBufferRemaining.getAndAdd(-size) > 0 ) {
322348
// just send it without blocking...
323349
connection.getDispatchQueue().execute(new Task() {
350+
@Override
324351
public void run() {
325352
connection.send(frame, new Callback<Void>(){
353+
@Override
326354
public void onFailure(Throwable value) {
327355
handleException(value);
328356
}
@@ -344,6 +372,7 @@ public void onSuccess(Void value) {
344372
}
345373
};
346374
connection.getDispatchQueue().execute(new Task() {
375+
@Override
347376
public void run() {
348377
connection.send(frame, future);
349378
}
@@ -359,6 +388,7 @@ public void run() {
359388

360389
public void sendRequest(final StompFrame frame, final Promise<StompFrame> future) {
361390
connection.getDispatchQueue().execute(new Task() {
391+
@Override
362392
public void run() {
363393
connection.request(frame, future);
364394
}
@@ -482,7 +512,7 @@ public URI getLocalURI() {
482512
public void setLocalURI(URI localURI) {
483513
this.localURI = localURI;
484514
}
485-
515+
486516
public boolean isOmitHost() {
487517
return omitHost;
488518
}

0 commit comments

Comments
 (0)