Skip to content

support stomp header for work queue and add support to subscribe active-mq temp-queue #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ellyxc wants to merge 1 commit into
base: master
Choose a base branch
from
Open

support stomp header for work queue and add support to subscribe active-mq temp-queue #21

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 26 additions & 8 deletions lib/adapters/stomp.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ var PREFIX = {
pull: 'queue',
pub: 'topic',
sub: 'topic',
tempPull: 'temp-queue'
};

function stompOpen(self, type, connection, name) {
Expand Down Expand Up @@ -143,15 +144,15 @@ function topicSelector(pattern) {
return 'topickey LIKE \''+pattern+'.%\' OR topickey = \''+pattern+'\'';
}

function stompSubscribe(self, pattern, callback) {
function stompSubscribe(self, pattern, callback, stompHeaders) {
var selector = topicSelector(pattern);

if (callback) {
self.on('message', callback);
}

c(self)._doWhenReady(function pullSubscribe(done) {
var headers = {};
var headers = stompHeaders || {};

dbg('pull subscribe start+done');

Expand Down Expand Up @@ -208,9 +209,9 @@ function decode(body, contentType) {
return JSON.parse(body);
}

function stompPublish(self, msg, topic) {
function stompPublish(self, msg, topic, stompHeaders) {
c(self)._doWhenReady(function pushPublish(done) {
var headers = {};
var headers = stompHeaders || {};

dbg(self.type+' publish start', msg);
var encoding = encode(msg);
Expand Down Expand Up @@ -241,8 +242,8 @@ function PushStomp(connection, name) {

util.inherits(PushStomp, events.EventEmitter);

PushStomp.prototype.publish = function(msg) {
return stompPublish(this, msg);
PushStomp.prototype.publish = function(msg, headers) {
return stompPublish(this, msg, null, headers);
};

PushStomp.prototype.close = stompClose;
Expand All @@ -257,8 +258,8 @@ function PullStomp(connection, name) {

util.inherits(PullStomp, events.EventEmitter);

PullStomp.prototype.subscribe = function(callback) {
return stompSubscribe(this, null, callback);
PullStomp.prototype.subscribe = function(callback, headers) {
return stompSubscribe(this, null, callback, headers);
};

PullStomp.prototype.close = stompClose;
Expand All @@ -267,6 +268,23 @@ CreateStomp.prototype.createPullQueue = function(name, callback) {
return new PullStomp(this, name, callback);
};

//-- Pull temp-queue

function PullTempStomp(connection, name) {
stompOpen(this, 'tempPull', connection, name);
}

util.inherits(PullTempStomp, events.EventEmitter);

PullTempStomp.prototype.subscribe = function(callback, headers) {
return stompSubscribe(this, null, callback, headers);
};

PullTempStomp.prototype.close = stompClose;

CreateStomp.prototype.createTempPullQueue = function(name, callback) {
return new PullTempStomp(this, name, callback);
};

//-- Pub/Sub Queue

Expand Down