smrchy / rsmq Goto Github PK

Thanks for this rsmq!

For my application, I send message with delay some seconds. Before it triggered in the future, I may change the message content or the delay time. Can you add this feature? Many thanks.

If node starts i don't know if the queue allready exists.
So i have to check via a .has() or cretae it o every init.

Possible solutions:

• a .has() Method to check
• a graceful answer of createQueue(). Possibly by an option
• add the queue-config to the constructor and create it internally if not exists.

Can we use this module on embedded devices? What are the pros and cons with that usage?

The queue name length limitation of 80 characters is a big issue for me. I create queues that are based on the value of a JWT token and so these exceed the 80 character limitation. Can this be overridden or just be made bigger?

Also the maximum message size of 65536 can be too small in some cases. Can the maximum be overridden or increased?

Constructor now accepts options instead of single args

what's your thinking on adding an event for when a queue gets a new message? thanks!

A protocol-document to implement rsmq within other languages would be quiet useful.

createQueue and setQueueParameters should support a maxsize=-1 option to allow unlimited size messages - at your own risk. Make sure Redis has enough RAM.

Would you accept a PR which allows you to send 2 or more messages atomically, i.e. it should send all messages, or if there is an error, none of them?

I'm getting the following error going on forever!

ReplyError: ERR unknown command 'time'
at parseError (/myproject/socket-server/node_modules/redis-parser/lib/parser.js:163:12)
at parseType (/myproject/socket-server/node_modules/redis-parser/lib/parser.js:224:14)

Hi,
First of all, many thanks for your great message queue implement using redis. Before known your lib, I try to use rabbitmq to setup message queue system for my app. But it's seem to be overkill and hard to maintain. Now I see your lib, It's seem to be very easy to use and redis server already running in my app. It's really cool.
So I wonder there is a way to use redis client on my rails app to push messages to my nodejs app directly?. I see you have made the rest-rsmq to easy communication between these apps. But I see it will hurt the performance when app have numerous concurrent request.
Looking forward to hearing from you, many thanks.

An option on createQueue and setQueueAttributes methods to set the time a message on a queue will be available to be received.

Errors should be returned as type Error not as a String

The title is self-explanatory, I think. I have the following pseudo-code:

rsmq.deleteMessage({qname:"myqueue", id:message.id}, function(err, resp){
    // the program could crash here and the job would never be finished
    rsmq.sendMessage({qname:"otherqueue", message:"finished-my-job"})
})

How to deal with these issues? I know this might not be directly related to RSMQ, I first thought "BRPOPLPUSH " from redis could help me somehow, but not sure how exactly.

What is the easiest way to handle the fact that an authentication password is required for redis and the fact that a specific database should be used?

The regex used to validate qname (line 532 of the built index.js file) does not match the allowed characters by from redis itself - colons ":" are allowed by redis and should be allowed by the package.

What do you think about an event wrapper around receiveMessage method? Then we can do something like this:

RedisSMQ = require("rsmq");
rsmq = new RedisSMQ( {host: "127.0.0.1", port: 6379, ns: "rsmq"} );

rsmq.createQueue({qname:"myqueue"}, function (err, resp) {
  if (resp === 1) {
    console.log("queue created")
  }
});

rsmq.onMessage({qname:"myqueue"}, function (err, resp) {
  if (resp.id) {
    console.log("Message received.", resp)  
  }
  else {
    console.log("No messages for me...")
  }
});

setInterval(function () {
  rsmq.sendMessage({qname:"myqueue", message:"Hello World"}, function (err, resp) {
    if (resp) {
      console.log("Message sent. ID:", resp);
    }
  });
}, 1000);

as published here mpneuried/rsmq-worker#21

Hi to all,

I'm using ioredis to establish a connection to Redis and use it as a client to RSMQ https://github.com/smrchy/rsmq and RSMQ-worker https://github.com/mpneuried/rsmq-worker.

If I use a standalone Redis server with default settings localhost:6370, all my code work as charm and tasks are being processed as expected.

Issues appear as soon as I try to use ioredis with Sentinels option (https://github.com/luin/ioredis#sentinel)

So after shut down the standalone Redis instance, I have set up a Sentinels environments with

a master server and 2 slaves servers in my local machine.
Then I connect to them with

var redis_client = new Redis({
     sentinels: [{host: "localhost", port: "16380"},{host: "localhost", port: "16381"},{host: "localhost",             port: "16382"}],
name: "redis-cluster"

});
as reported in https://github.com/luin/ioredis#sentinel.
Apparently everything seems to work correctly, and

redis_client.on('connect;,function(){}
says that the client has been connected to the Sentinels servers.
Then according to the RSMQ documentation I can use the already existing Redis client with the RSMQWorker

var worker = new RSMQWorker("mytask", {
interval: [0, 10],
autostart: true,
maxReceiveCount: variables.rsmq.maxReceiveCount,
customExceedCheck: fnCheck,
redis: redis_client,
redisPrefix: "rsmq",
alwaysLogErrors: true
});
but unfortunately with these settings tasks stop to work.
As a clue the Redis Master server receives nothing on monitor.
So I started to investigate and first of all I have printed out the redis_client object, just to discover a couple of localhost connections in it.

[...]
options:
{ sentinels: [ [Object], [Object], [Object], [Object], [Object] ],
name: 'redis-cluster',
port: 6379, <---------
host: 'localhost', <----------
[...]
SentinelConnector {
options:
{ sentinels: [Object],
name: 'redis-cluster',
port: 6379, <------
host: 'localhost', <------
family: 4,
[...]
Does someone knows why even if I connect using Sentinels, the redis_client json has localhost and local port values on ?

I was wondering if someone has experience with Sentinels and with these kind of issues.

Thanks in advance

REDIS CLIENT: Redis {
domain: null,
_events:
{ connect: [ [Function], [Object] ],
close: { [Function: g] listener: [Function] } },
_eventsCount: 2,
_maxListeners: undefined,
options:
{ sentinels: [ [Object], [Object], [Object], [Object], [Object] ],
name: 'redis-cluster',
port: 6379,
host: 'localhost',
family: 4,
connectTimeout: 3000,
retryStrategy: [Function],
keepAlive: 0,
connectionName: null,
role: 'master',
sentinelRetryStrategy: [Function],
password: null,
db: 0,
parser: null,
dropBufferSupport: false,
enableOfflineQueue: true,
enableReadyCheck: true,
autoResubscribe: true,
autoResendUnfulfilledCommands: true,
lazyConnect: false,
keyPrefix: '',
reconnectOnError: null,
readOnly: false,
stringNumbers: false },
scriptsSet: {},
commandQueue: { [String: '[object Object]'] _capacity: 16, _length: 1, _front: 0 },
offlineQueue: { [String: ''] _capacity: 16, _length: 0, _front: 0 },
connector:
SentinelConnector {
options:
{ sentinels: [Object],
name: 'redis-cluster',
port: 6379,
host: 'localhost',
family: 4,
connectTimeout: 3000,
retryStrategy: [Function],
keepAlive: 0,
connectionName: null,
role: 'master',
sentinelRetryStrategy: [Function],
password: null,
db: 0,
parser: null,
dropBufferSupport: false,
enableOfflineQueue: true,
enableReadyCheck: true,
autoResubscribe: true,
autoResendUnfulfilledCommands: true,
lazyConnect: false,
keyPrefix: '',
reconnectOnError: null,
readOnly: false,
stringNumbers: false },
connecting: true,
retryAttempts: 0,
currentPoint: 0,
sentinels: [ [Object], [Object], [Object], [Object], [Object] ],
stream:
Socket {
_connecting: false,
_hadError: false,
_handle: [Object],
_parent: null,
_host: null,
_readableState: [Object],
readable: true,
domain: null,
_events: [Object],
_eventsCount: 7,
_maxListeners: undefined,
_writableState: [Object],
writable: true,
allowHalfOpen: false,
destroyed: false,
bytesRead: 0,
_bytesDispatched: 14,
_sockname: null,
_pendingData: null,
_pendingEncoding: '',
server: null,
_server: null,
_idleTimeout: -1,
_idleNext: null,
_idlePrev: null,
_idleStart: 1244,
read: [Function],
_consuming: true,
_peername: [Object] } },
retryAttempts: 0,
status: 'connect',
condition: { select: 0, auth: null, subscriber: false },
stream:
Socket {
_connecting: false,
_hadError: false,
_handle:
TCP {
_externalStream: {},
fd: 16,
reading: true,
owner: [Circular],
onread: [Function: onread],
onconnection: null,
writeQueueSize: 0 },
_parent: null,
_host: null,
_readableState:
ReadableState {
objectMode: false,
highWaterMark: 16384,
buffer: [],
length: 0,
pipes: null,
pipesCount: 0,
flowing: true,
ended: false,
endEmitted: false,
reading: true,
sync: false,
needReadable: true,
emittedReadable: false,
readableListening: false,
defaultEncoding: 'utf8',
ranOut: false,
awaitDrain: 0,
readingMore: false,
decoder: null,
encoding: null,
resumeScheduled: false },
readable: true,
domain: null,
_events:
{ end: [Object],
finish: [Function: onSocketFinish],
_socketEnd: [Function: onSocketEnd],
error: [Object],
close: [Object],
data: [Function],
timeout: [Object] },
_eventsCount: 7,
_maxListeners: undefined,
_writableState:
WritableState {
objectMode: false,
highWaterMark: 16384,
needDrain: false,
ending: false,
ended: false,
finished: false,
decodeStrings: false,
defaultEncoding: 'utf8',
length: 0,
writing: false,
corked: 0,
sync: false,
bufferProcessing: false,
onwrite: [Function],
writecb: null,
writelen: 0,
bufferedRequest: null,
lastBufferedRequest: null,
pendingcb: 1,
prefinished: false,
errorEmitted: false,
bufferedRequestCount: 0,
corkedRequestsFree: [Object] },
writable: true,
allowHalfOpen: false,
destroyed: false,
bytesRead: 0,
_bytesDispatched: 14,
_sockname: null,
_pendingData: null,
_pendingEncoding: '',
server: null,
_server: null,
_idleTimeout: -1,
_idleNext: null,
_idlePrev: null,
_idleStart: 1244,
read: [Function],
_consuming: true,
_peername: { address: '127.0.0.1', family: 'IPv4', port: 6380 } },
replyParser:
JavascriptReplyParser {
name: 'javascript',
buffer: ,
offset: 0,
bigStrSize: 0,
chunksSize: 0,
buffers: [],
type: 0,
protocolError: false,
offsetCache: 0,
handleReply: [Function],
handleNumbers: [Function],
returnError: [Function],
returnFatalError: [Function],
returnReply: [Function] } }

Hi,

My environment is:

OS: Ubuntu 16.04.01
Kernel: 4.4.0-75-generic
Node: 4.2.3
Redis Server: 3.2.5
rsmq: 0.8.2
rsmq-worker: 0.5.2

I have not made any changes to the library source. The queuing performs consistently satisfactorily:

[ RSMQ ] Time Taken:  28
[ RSMQ ] Time Taken:  3
[ SQS ] Time Taken:  303
[ SQS ] Time Taken:  214
[ RSMQ ] Time Taken:  3
[ RSMQ ] Time Taken:  3
[ RSMQ ] Time Taken:  3
[ SQS ] Time Taken:  628
[ SQS ] Time Taken:  455
[ SQS ] Time Taken:  207
[ RSMQ ] Time Taken:  4
[ SQS ] Time Taken:  214
[ RSMQ ] Time Taken:  4
[ SQS ] Time Taken:  192
[ RSMQ ] Time Taken:  6
[ SQS ] Time Taken:  244
[ RSMQ ] Time Taken:  5
[ SQS ] Time Taken:  198
[ RSMQ ] Time Taken:  4
[ SQS ] Time Taken:  184

However, the event takes too long to reach the worker.

Though the Redis server is local, rsmq-worker takes an awfully long time to receive the "data" event and process it.

var RSMQWorker = require( "rsmq-worker" );
var options = 
{
  invisibletime: 900,                    // hide received message for 900 sec
  maxReceiveCount: 1,              // only receive a message once until delete
  autostart: true,                         // start worker on init
  customExceedCheck: exdCheckFn
};
var worker = new RSMQWorker( "testqueue", options);
worker.on( "data", function( msg ){
 // do stuff
});

For a set of 10 messages that were set in SQS and redis at the same time:

[ SQS ] Total Time taken for processing at site 13
[ SQS ] Total Time taken for processing at site 14
[ SQS ] Total Time taken for processing at site 38
[ SQS ] Total Time taken for processing at site 27
[ SQS ] Total Time taken for processing at site 139
[ SQS ] Total Time taken for processing at site 20
[ SQS ] Total Time taken for processing at site 455
[ SQS ] Total Time taken for processing at site 416
[ SQS ] Total Time taken for processing at site 1375
[ SQS ] Total Time taken for processing at site 125
[ RSMQ ] Total Time taken for processing at site 3428
[ RSMQ ] Total Time taken for processing at site 6190
[ RSMQ ] Total Time taken for processing at site 8980
[ RSMQ ] Total Time taken for processing at site 11767
[ RSMQ ] Total Time taken for processing at site 14532
[ RSMQ ] Total Time taken for processing at site 17331
[ RSMQ ] Total Time taken for processing at site 20097
[ RSMQ ] Total Time taken for processing at site 22852
[ RSMQ ] Total Time taken for processing at site 25590
[ RSMQ ] Total Time taken for processing at site 28347

As you can see even AWS SQS is outperforming it. Are there any obvious settings that I am misconfiguring?

Whether the future version supports message priority settings?

If you set a prefix value on the redis client everything starts to break. If you change line 15 of the unit test file from this:

redis = RedisInst.createClient();

to this:

redis = RedisInst.createClient(6379, '127.0.0.1', {prefix: 'someprefix'});

Unit tests will begin to fail.

Hey,

What happens if a queue already exists?

greets
Philipp

First of all thank you for rsmq, it's an amazingly simple message queue solution. Due to it's simplicity it's a good way to share a message queue layer in different technologies.

In our case, Java was sending messages to Node.js in the same infrastructure, therefore we decided to implement (port) this project. It's available here https://github.com/wedeploy/jrsmq. We've tried to keep the api as similar as possible. If you are interested we can support it as the official Java version.

Just getting my feet wet with your lovely library. Perhaps I have a fundamental misunderstanding but is it possible to have more than one instance of RSMQ populating the same queues?

I start redis in standalone mode so it's running at 6379 default port

npm install rsmq --save
npm -i (to get into console)
RedisSMQ = require("rsmq");
rsmq = new RedisSMQ( {host: "127.0.0.1", port: 6379, ns: "rsmq"} );

I don't see anything in redis, no logging about connection

then rsmq.connected returns true, but any attempts to use further gives me errors like this

such as, after using the createQueue like showin the in the examples, I run

> rsmq.listQueues()
undefined
> TypeError: cb is not a function

I want to receive message from several queues using the same method, can I subscribe using a wildcard? (The issue is that my application can support many different pipelines each of which has its own queue)

I apologize for my lack of knowledge surrounding coffeescript, but I often prefer a call syntax for modules that take parameters...

var rsmq = require('rsmq')(options);

... over invocation of an instance... If this syntax is supported, It would be nice to see it in the documentation, if it isn't it would be a nice addition.

In various flavors of AMQP it is possible to specify a special queue to route messages into after a max number of retry attempts has been exceeded or if the message is undeliverable after a timeout expiration. Something similar for RSMQ could be valuable to have for debugging.

I know I could implement something similar myself using the "exceeded" event on a worker but what if no workers are online? Also it isn't clear whether that exceeded count is shared between workers or specific to only that instance.

Error: [invalidFormat: Invalid <%= item %> format]

<%= item %> is not replaced

I wonder how to create delayed message queue. Please take a look on my test case and the result

it('should create send messagee', function (done) {
    rsmq.deleteQueue({'qname': "myqueue"}, function (res) {
      rsmq.createQueue({qname: "myqueue"}, function (err, resp) {
        if (resp === 1 || err.name === 'queueExists') {
          console.log("queue created");
          rsmq.sendMessage({qname: "myqueue", message: "Hello world", "delay": 1}, function (err, resp) {
            if (resp) {
              console.log("message sent");
              console.log("Message sent. ID:" + resp);
              done();
            }
          });
        }
      });
    });
  });

  it('should receive msg', function (done) {
    rsmq.receiveMessage({qname: "myqueue"}, function (err, resp) {
      if (resp.id) {
        console.log("Message received.")
        console.log(resp);
      }
      else {
        console.log(err);
        console.log("No messages for me...");
      }
    });
  });

Result:
queue created
message sent
Message sent. ID:e2mp1o4p90fNowqfWpNbxNdAj2huPIKN
.null
No messages for me...

I didn't put done inside receiveMessage function and wait until timeout but the receiveMessage never triggered again.

The generated ids from receiveMessage will not validate

Test Script:

_makeid = (len) ->
    text = ""
    possible = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
    for i in [0...len]
        text += possible.charAt(Math.floor(Math.random() * possible.length))
    return text

uid = _makeid(22)

r = /^([a-zA-Z0-9:]){32}$/
realId = "dj24w17ku593c0b83d5c913def307e6e845eef371a"

simId = Number(Date.now() + "0000").toString(36) + uid

[r.test( realId ),r.test( simId ), "dj24w17ku593c0b83d5c913def307e6e845eef371a".length ]

Hey guys.

Is it possible to, instead of waiting the visibility timeout, send the message back to the queue immediately?

Would "changeMessageVisibility" help me in this? After a message is received, can I call "changeMessageVisibility" with vt 0 to push it back?

Let's say you have a client and server situation. The client can enqueue messages and the server can pull the messages off the queue. Can you define the server so that it will "wait" for a message to arrive so that it can then retrieve it and then go back and wait for more?

The documentation seems to imply that you should only retrieve a message if you know one is there, because a retrieve will return nothing if there are no messages, rather than wait for something to arrive - but that doesn't make much sense :(

There seems to be a missing "consume" method which would allow a process to wait for messages to arrive, and retrieve them when they do.

This library looks really interesting, but I wonder why native pub/sub functions of Redis is not used, instead library is utilizes own implementation.

Was it created before pub/sub appeared in Redis?

Hi, is there any way to update existing message?

Let's say, I queued a message, then worker received it and when tried to process it, it failed, thus I wanna set a 'mark' on that job to act in a different manner (or just update existing message, whatever).

Is there any way to do that or do you plan to add such thing?

Hello Patrick

What do you think of adding a rate limit (throttling) on received messages in flight?

For example, Amazon SES has a maximum limit for emails sent per seconds. A typical worker could over send if the queue messages are over the limit. The same applies when you call an API and don't want to overflow it.

In Redis documentation, there is an example of applying rate limit: http://redis.io/commands/INCR#pattern-rate-limiter-1

Cheers

HI,

Thanks for the module.

I have a issue, I sent message and I got the message ID too. But, the receiver not received any message.

Thanks,
Sabari.

I looked at the code, but couldn't understand. Redis does not have this "delay before add" feature out-of-the-box, does it? How do you do it?

Looking at the constructor, it seems that if a Redis instance is passed in, and it has already connected, rsmq won't call the handler for the @redis.on "connect" event.

This means that the lua scripts won't be initialised in this case?

Any plans to support the broadcasting of messages?

It would be nice if you added a promisified implementation...

var rsmq = require('rsmq')(options).promise;

...

rsmq.receiveMessage(...)
  .then(function(message){
    ...
  })
  ...
  .catch(function(err){
  });

You could either shim in promises via i-promise or use something like bluebird directly.

const RedisSMQ = require("rsmq")
const Redis = require("ioredis")

const redis = new Redis("redis://:[email protected]:6379/0")
const rsmq = new RedisSMQ({ client: redis })

rsmq.sendMessage({ qname: queue_name, message: JSON.stringify(message) }, function (err, resp) {
	console.log(err, resp)
})

All connected, but sendMessage callback not firing, if I'm disable auth in redis config, get rid of ":foobared" all works.

i want using redis sentinel but i found snippets source code

if opts.client?.constructor?.name is "RedisClient"
			@redis = opts.client
		else
			@redis = RedisInst.createClient(opts.port, opts.host, opts.options)

node-redis-sentinel/ ioredis constructor.name is "RedisSentinelClient"/"Redis"

node-redis-sentinel-client
ioredis

hi, I looked up at rsmq and I liked it a lot! (tried kue and bull but for some reasons, they add too much overhead and got some bugs which is not acceptable for me).

The only thing I'm interested at the moment is that, why not use JSON.stringify/parse (or MessagePack) directly instead of checking typeof message and throwing error? I really want to send objects instead of just strings, but writing it manually by myself ain't a great thing though.

Here: https://github.com/smrchy/rsmq/blob/master/index.js#L475

you should use hset instead of hsetnx, as the second one does not have any effect if hash already exist.

Thanks for a great queue library!

For my application, I'm looking to delete a message as it is received. I can currently do this by issuing a deleteMessage straight after a receiveMessage, but I'd like to go one step further and add support for this in the receiveMessage lua script, so that the delete happens atomically.

I was wondering whether you'd accept a PR which implements this functionality.

Here's what I'm thinking:

• Allow a vt of -1 - this indicates that a message should be deleted when it is pulled off the queue.
• Check if vt is -1 in the lua script - if so, run the zrem and hdel operations to delete the message.

If reusing the vt variable isn't desirable, perhaps a new option could be used instead, for example remove_on_receive.

Hi there.
First I wanna thank you for your project, it is very helpful to me.

I have a question, I didn't find the answer in the documentation. Is there a way to send a message with a expiration time, I mean, if there is no client listening for the message after after a period of time the message expires.

I think I can implement some kind of timeout that calls deleteMessage() after a x period of time but I have no control if a message was send to a client before I deleted.

Regards.

Hi,
Is it possible to use Ioredis instead of Hiredis as a Redis driver?
With the current driver we can't do high availability.

Hitting message too long as defined in https://github.com/smrchy/rsmq/blob/deb56ab82f6349b9578bc8d4e15286ca53fa058f/test/test.coffee (could't find other place it's defined in code)

It this a redis imposed threshold? Do you happen to know the max size?

For now I can say it doesn't seem to work with 2.2.2