Hi,

We are running in a redundant redis setup with sentinels. When I kill the master redis(kill the VM), it takes a while for a redundant master to take over. During that time, I guess an error is sent by ioredis and, since that error is thrown by the redis-client, it never gets caught.
Is there a reason why there is no recovery after a transient redis error
this.client.once('error', (err: Error) => {
throw err;
});

Thanks

When a connection error occurs (Connection timetout), one of the events will end the main application and there is no way to prevent that.
There should be some handler that allowed whoever used the application to handle the failure and decide by himself if the application should be terminated.

File:

• node_modules/redis-smq/src/dispatcher.js

.on(events.ERROR, (err) => {
                if (err.name !== 'AbortError' && state !== states.GOING_DOWN) {
                    loggerInstance.error(err);
                    process.exit(1);
                }
            });

events.js:183
      throw er; // Unhandled 'error' event
      ^
ReplyError: ERR wrong number of arguments for 'set' command
    at parseError (C:\Users\212702834\Documents\Userdata\Practice\redis-message-queue\node_modules\redis-parser\lib\parser.js:193:12)
    at parseType (C:\Users\212702834\Documents\Userdata\Practice\redis-message-queue\node_modules\redis-parser\lib\parser.js:303:14)

Dependency Versions:

{
    "redis": "2.7.1",
    "redis-smq": "^1.0.20"
  }

Running on Windows 10, x64
Redis server version 2.4.5

AbortError: Connection forcefully ended and command aborted. It might have been processed.
at RedisClient.flush_and_error (/data/site/quan-share/node_modules/redis/index.js:357:23)
at RedisClient.end (/data/site/quan-share/node_modules/redis/lib/extendedApi.js:51:14)
at handler (/data/site/quan-share/node_modules/redis-smq/src/scheduler.js:180:24)
at Object.stop (/data/site/quan-share/node_modules/redis-smq/src/scheduler.js:193:20)
at Object.shutdown (/data/site/quan-share/node_modules/redis-smq/src/dispatcher.js:390:27)
at Producer.shutdown (/data/site/quan-share/node_modules/redis-smq/src/producer.js:62:25)
at producer.produce (/data/site/quan-share/utils/log.js:135:16)
at Command.redisClientInstance.lpush [as callback] (/data/site/quan-share/node_modules/redis-smq/src/dispatcher.js:431:25)
at normal_reply (/data/site/quan-share/node_modules/redis/index.js:721:21)
at RedisClient.return_reply (/data/site/quan-share/node_modules/redis/index.js:819:9)
at JavascriptRedisParser.returnReply (/data/site/quan-share/node_modules/redis/index.js:192:18)
at JavascriptRedisParser.execute (/data/site/quan-share/node_modules/redis-parser/lib/parser.js:574:12)
at Socket. (/data/site/quan-share/node_modules/redis/index.js:274:27)
at emitOne (events.js:116:13)
at Socket.emit (events.js:211:7)
at addChunk (_stream_readable.js:263:12)
at readableAddChunk (_stream_readable.js:250:11)
at Socket.Readable.push (_stream_readable.js:208:10)
at TCP.onread (net.js:607:20)

I'm trying to figure out how to add delay between retry on error. Neither setScheduledPeriod or
setScheduledDelay seems to do it. I want to retry a .job n number of times only on failure and there must be 10 minutes delay between retry. Is this possible to do?

Message well send to all running Consumer when the function consume call cb(err) on some Consumer.
Is there bug.

Hello admin,
I had a problem with stats.js on line 78. I think need remove comma at the end of the line 77.

D:\server\chat-server\node_modules\redis-smq\src\stats.js:78
);
^
SyntaxError: Unexpected token )
at createScript (vm.js:53:10)
at Object.runInThisContext (vm.js:95:10)
at Module._compile (module.js:543:28)
at Object.Module._extensions..js (module.js:580:10)
at Module.load (module.js:488:32)
at tryModuleLoad (module.js:447:12)
at Function.Module._load (module.js:439:3)
at Module.require (module.js:498:17)
at require (internal/module.js:20:19)
at Object. (D:\server\chat-server\node_modules\redis-smq\src\consumer.js:8:22)

Are they available? I would like to try this package in my TS app.

Hi!

My dead-letter queue (DLQ) is full!
So, I fixed the error!!
And now, how can I move dead-letter queue to main queue for consumer process the message again?

In the consumer constructor, project namespace must be set (setNamespace()) before generating the keys (redisKeys.getKeys()).

Currently, while construing consumer this is causing the keys to be generated incorrectly with default namespace in it, like redis-smp-default. Rest of the times, getKeys generating correct keys.

Monitor needs namespace update in statsFrontend()
if (config.hasOwnProperty('namespace')) redisKeys.setNamespace(config.namespace);

if redis server doesn't start, run monitorServer.listen() will get app crash

The producer cannot check the existence of the message on restart and duplicates it.
There is no way to set a unique name for a cron task and update it instead of creating duplicates.
There is no way to delete a task that runs regularly.
Thus, the use of cron tasks loses all meaning, without being able to manage them and control their work.

Is it possible?

thanks

@weyoss Thanks for nicely crafted library.

I have done some performance benchmark to compare redis-smq with bull on my Dell XPS. Machine 16GB RAM, Hexa Core

No doubt, redis-smq is clear winner because of it's simplicity and focused approach towards message queuing.

Following is the messages produced and consumed rate

It is clearly visible that bull is almost saturated in terms of producer and consumer rate.
There can be multiple reasons of less performance in bull like promises or multiple checks & conditions to provide other features like rate limiting etc.

Though, I am not able to convince myself which library to use for my next project, because redis-smq simplicity and performance is far ahead of bull, which is the purpose of this lib, but bull has lot of features, which can be useful when things are in production.

I am thinking to move from RabbitMQ to Redis, because I don't need complex routing like RabbitMQ.
I have also seen heavy usage for simple queuing task in RabbitMQ, where RabbitMQ was taking approx 2GB for holding 1 million messages and total messages size was around 100MB. So ideally in Redis, it would be 100MB only. I understand RabbitMQ do lot of things out of the box, but I think Redis can work very well in simple queuing cases.

Any suggestion would be appreciated by anyone to take this decision. I am also interested to listen more from actual user of this library who have used it in production with heavy load.

Thanks.

hi,
what is the best way to check if the que is not empty? should i create a socketio client like the one in monitor GUI?

Steps to reproduce:

1. Send message to one Queue1
2. Consumer1 crashes due to error
3. Both Queue1 and Queue2 receive the message from Step1

ps: redis is running with aof

Hi there,
In my use case, I have to send the same message to a large number of queues. By using Producer class, I will have to create multiple instances of producers for multiple queues.

What's the more efficient way to do this? e.g. can we have something like producer.produce(queue, message, cb)?

Thanks.

Need to stop the producer without execute "process.exit(0);"

consumer.js
In line 220, its should be message.uuid instead of message.id
this[sLogger].error(Consumer failed to consume message [${message.id}]...);

so minor :)

Currently, redis password is not used while creating redis connection. Need to support redis password and db.

Please suggest, should the getNewInstance (redis-client.js) be modified to include password and db or is there a better way to do it without touching this library files.

Thank you.

I am trying to create alerts and/or display queue size in our internal monitoring/alerting/dashboard system. Is there any way to get the metrics from the queues other than through the monitoring web site?

Let's say we have some items which are higher priority (for example, because they are fast to process).

Is there a recommended way to enqueue them at or near the front of the queue?

Hi.

Does your app have or will have support for a clustered redis database?

Keep up the good work.

First of all I must say I'm super happy with the performance of this Plugin. I would like to make an open question regarding redis-smq.

In the admin dashboard I'm working on, I'd like to have the hability of displaying the amount of running tasks, failed queues, etc.

I tried with a simple redis client, and trying to search on the following keys:

const keyTypes = {
    KEY_TYPE_MESSAGE_QUEUE: '1.1',
    KEY_TYPE_PROCESSING_QUEUE: '1.2',
    KEY_TYPE_DEAD_LETTER_QUEUE: '1.3',
    KEY_TYPE_MESSAGE_QUEUE_DELAYED: '1.4',
    KEY_TYPE_HEARTBEAT: '2.1',
    KEY_TYPE_GC_LOCK: '3.1',
    KEY_TYPE_GC_LOCK_TMP: '3.2',
    KEY_TYPE_RATE: '4',
    KEY_TYPE_RATE_INPUT: '4.1',
    KEY_TYPE_RATE_PROCESSING: '4.2',
    KEY_TYPE_RATE_ACKNOWLEDGED: '4.3',
    KEY_TYPE_RATE_UNACKNOWLEDGED: '4.4',
    KEY_TYPE_STATS_FRONTEND_LOCK: '5.1',
    KEY_TYPE_MESSAGE_QUEUES_INDEX: '6.1',
    KEY_TYPE_PROCESSING_QUEUES_INDEX: '6.2',
    KEY_TYPE_DEAD_LETTER_QUEUES_INDEX: '6.3',
    KEY_TYPE_SCHEDULER_LOCK: '7.1',
    KEY_TYPE_SCHEDULER_LOCK_TMP: '7.2',
};

Ideally I would like to understand how to search in this tree of keys, and avoid using a redis client. I could implement that part perfectly and would be happy to help but i'm lost on how finding the info.

If we are talking about high performance, then a large number of connections create locks at the redis level, and in our case, these are 15 connections per 1 Producer and 1 consumer:
Consumer - 11 connections
Producer - 4 connections

I researched the code and saw no reason to use that many connections.
For a producer, 1 connection is enough.
For the consumer, 3 connections are enough, one for the main process and two for the forks.

Highly optimized: No promises, no async/await, small memory footprint, no memory leaks. See callbacks vs promises vs async/await benchmarks.

@weyoss i assume this statement is out of date. And the event loop has been optimized to work with async/await.
Please, check this benchmark

Hi,
I have recently come across this module.
Does it allow setting message-id to a job like this https://github.com/bee-queue/bee-queue#jobsetidid

I tried bee-queue but it has some bugs which are not acceptable in our case.

Also, can I provide concurrency to consumers?

How do I define the db function? Does this work?

const redisSMQ = require('redis-smq');
const ingestion = require('./ingestion');

const { Consumer } = redisSMQ;

class QueueConsumer extends Consumer {
  // eslint-disable-next-line class-methods-use-this
  consume(message, cb) {
    ingestion.myfunction(message);
    cb();
  }
}

module.exports = { QueueConsumer };

And also how do we check if the queue is empty?

Hi,

If a consumer throws a message before sending the acknowledgment then it is received after the consumer is restarted. But if the retry attempt is more than 3 then the message is deleted. What should be the value of setRetryThreshold so that the message is not automatically deleted unless the consumer sends ACK? Have tried setting message.setRetryThreshold(-1) but it doesn't work. Also, what is the default value of messageConsumeTimeout?

//Producer.ts
'use strict';
const { Message, Producer } = require('redis-smq');

const message = new Message();

message
    .setBody({hello: 'worldd!'})
    .setTTL(10800000)

const config = {
    namespace: 'test',
    redis: {
        driver: 'redis',
        options: {
            host: '127.0.0.1',
            port: 6379,
            connect_timeout: 3600000,
        },
    }
}
const producer = new Producer('test_queue',config); //test_queue
producer.produceMessage(message, (err) => {
   if (err) console.log(err);
   else console.log('Successfully produced')
});

//Consumer.ts
'use strict';

const redisSMQ = require('redis-smq');
const Consumer = redisSMQ.Consumer;

class TestQueueConsumer extends Consumer {
    constructor() {
        super(config)
    }
    consume(message, cb) {
        try{
            console.log('Got a message to consume:', message);
            throw new Error("test")
            //cb();
        } catch(e) {
            console.log("consumer error",e)
        }
        
    }
}

TestQueueConsumer.queueName = 'test_queue';
const consumer = new TestQueueConsumer();
consumer.run();

/home/aappana/nodejs/node_modules/redis-smq/src/monitor/index.js:33
app.use(async (ctx) => {
^

SyntaxError: Unexpected token (
at createScript (vm.js:56:10)
at Object.runInThisContext (vm.js:97:10)
at Module._compile (module.js:542:28)
at Object.Module._extensions..js (module.js:579:10)
at Module.load (module.js:487:32)
at tryModuleLoad (module.js:446:12)
at Function.Module._load (module.js:438:3)
at Module.require (module.js:497:17)
at require (internal/module.js:20:19)
at Object. (/home/aappana/nodejs/node_modules/redis-smq/index.js:5:17)

Ran this in linux - As this library doesnot support windows. your project architecture was good. but failed.

Is there some support for recurring tasks, similar to Bull? I suspect I could use something like later, but it would be good to know...

Hi,

Is it possible to create new consumers on the fly during a running node process rather than have them run separately as a standalone CLI.
Excellent work on creating this SMQ.

Thanks,
Mahesh

It's not explicitly stated but the library appears to rely on Redis being configured for persistence. In other words: there is no way for us to provide our own persistence mechanism.

Our product is deployed to a large number of customers with Redis persistence disabled. I would LOVE to be able to just switch Redis persistence on, but updating deployments is... complicated and likely to break installations. I have reached out to other contributors and management for input on making this switch but, in the mean time, I'm looking at alternatives.

Would a PR that allows us to provide a "Persistence" class be accepted? The assumption is that the default class's functionality would always be a no-op and rely on Redis. In our case, we could provide an instance of this class through the configuration and the library would call the functions on that instance to persistence the data else-where.

In a lot of projects there are existing loggers. It'd be great to accept a bunyan compatible logger instance in the options and use that over making one internally. This will allow the host app to use their own levels and transports without duplicating those over to redis-mq.

Instead of shutting down the producer if there is an elegant way to call a function and stop the published message that will be nice. This could be useful if the publish is triggered by the user action from the client side and can be useful stopping the previous publish and generate new one.

Since upgrading to the latest version(2.0.11), I always get this error when I failover my redis server.
Error: Can not switch to the same current state.
at switchState (/opt/app-root/src/node_modules/redis-smq/src/power-state-manager.js:19:19)
at Object.goingUp (/opt/app-root/src/node_modules/redis-smq/src/power-state-manager.js:62:13)
at Object.start (/opt/app-root/src/node_modules/redis-smq/src/scheduler.js:219:31)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:46:36)
at Producer.emit (events.js:314:20)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:43:18)
at Producer.emit (events.js:314:20)
at Producer.completeBootstrap (/opt/app-root/src/node_modules/redis-smq/src/instance.js:264:14)
at Producer. (/opt/app-root/src/node_modules/redis-smq/src/instance.js:40:60)
at Producer.emit (events.js:314:20)
at /opt/app-root/src/node_modules/redis-smq/src/instance.js:195:23
at /opt/app-root/src/node_modules/neo-async/async.js:6883:13
at /opt/app-root/src/node_modules/redis-smq/src/instance.js:184:22
at tryCatcher (/opt/app-root/src/node_modules/standard-as-callback/built/utils.js:11:23)
at /opt/app-root/src/node_modules/standard-as-callback/built/index.js:19:49
at processTicksAndRejections (internal/process/task_queues.js:93:5)

Looks like the GOING_UP in the power-state-manager is emitted while it is already UP. I am using ioredis with sentinel support, the failover takes ~10sec.
I haven't seen this issue with version 2.0.0

npm ERR! Windows_NT 10.0.15063
npm ERR! argv "C:\Program Files\nodejs\node.exe" "C:\Program Files\nodejs\node_modules\npm\bin\npm-cli.js" "install" "redis-smq" "--save"
npm ERR! node v7.0.0
npm ERR! npm v3.10.8
npm ERR! code ENOSELF

npm ERR! Refusing to install redis-smq as a dependency of itself
npm ERR!
npm ERR! If you need help, you may report this error at:
npm ERR! https://github.com/npm/npm/issues

npm ERR! Please include the following file with any support request:
npm ERR! C:\xxxxxxx\examples\nodejs\redis-smq\npm-debug.log

node -v = 7.0.0

Typescript can't compile because of an error.
In types/config.ts interface ConfigInterface extends Monitor.ConfigInterface

export interface ConfigInterface extends Monitor.ConfigInterface {
    namespace?: string,
    redis: {
        driver: ConfigRedisDriver,
        options: {
            [key:string]: any
        }
    },
    log?: {
        enabled: boolean,
        options: {
            [key:string]: any
        }
    },
}

but redis is not suitable for Monitor.ConfigInterface interface, because of different types of driver. They look equal, but that's different enums.

export enum ConfigRedisDriver {
    REDIS = "redis",
    IOREDIS = "ioredis"
}

and

export enum RedisDriver {
    REDIS = 'redis',
    IOREDIS = 'ioredis'
}

In version 2.0.0 there wasn't inheritance and there wasn't a typescript error. In 2.0.1 it was added.

It would be helpful to have a configurable namespace instead of the default hard coded ns redis-smq, via the config file. Helpful in cases where redis-smq is a module of a larger redis based projects.

Hi,

After researching for all the available Redis queues, I found this to be the best for our use case :)

Since I haven't encountered any errors yet and it fulfills most of the use case.

I just wanted to clarify a few things, Let me know if this is not possible.
-I have multiple workers that process jobs, no 2 workers should process the same job.
-After a worker's job is complete the worker ACKs the job ( using cb() ). Until then the job is unavailable for other workers.
-If one of the apps terminates abruptly; e.g. via CTRL-C then the job should be sent back to the queue so that it can be picked by the next available worker. Which I think it does.
(correct me if it will not since that is the most important criteria)

However, I do not want to remove the message automatically from the queue. Messages should only be removed by the worker.
I tried doing this message.setBody(msg).setTTL(-1) but it gives error.

Is there a way to do so. If not what is the maximum TTL that I can provide.

My consumer class looks like:

(tried with with and without the constructor method. Same issue)

import {Consumer} from 'redis-smq';

/**
 * Consumer manager
 */
export default class ConsumerQueueManager extends Consumer {
  constructor() {
    super();
  }

  /**
   * @param {Message} message
   * @param {function} cb
   */
  consume(message, cb) {
    console.debug(message);
    cb();
  }
}

my application kernel (who starts the consumer)

/**
   * Run consumer payment process
   */
  async runConsumerQueueProcess() {
    if (!process.argv.includes('--process-consumer-queue-disabled')) {
      this._logger.info('Starting consumer queue process');
      ConsumerQueueManager.queueName = config.queue.name;
      const consumerManager = new ConsumerQueueManager();
      consumerManager.run();
    }
  }

got this error:

error: [general] TypeError: Class constructor Consumer cannot be invoked without 'new'

Was the ns intended to be redis-smp- instead of redis-smq ?

Hi there, with Producer, I'm able to produce/consume the message, but with MultiQueueProducer, it seems it doesn't work.

This works

const { Consumer, Producer, Message } = require('redis-smq');
const queue = 'S-12345'

const message = new Message();
message
    .setBody({ hello: 'world', ts: new Date() })
    .setTTL(3600000);

const producer = new Producer(queue);

setInterval(() => {
    producer.produce(message, (err) => {
        if (err) console.log(err);
        else console.log('Successfully produced')
    });
}, 1000);

class TestQueueConsumer extends Consumer {
    consume(message, cb) {
        console.log('Got a message to consume:', message.body);
        cb();
    }
}

const consumer = new TestQueueConsumer(queue);
consumer.run();
console.log('consumer running')

This doesn't work

const { Consumer, MultiQueueProducer, Message } = require('redis-smq');

const queue = 'S-12345'

const message = new Message();
message
    .setBody({ hello: 'world', ts: new Date() })
    .setTTL(3600000);

const producer = new MultiQueueProducer();

setInterval(() => {
    producer.produce(queue, message, (err) => {
        if (err) console.log(err);
        else console.log('Successfully produced')
    });
}, 1000);

class TestQueueConsumer extends Consumer {
    consume(message, cb) {
        console.log('Got a message to consume:', message.body);
        cb();
    }
}

const consumer = new TestQueueConsumer(queue);
consumer.run();
console.log('consumer running')

Anything wrong? Thanks in advance.

Hi there.

While consumer is running, when I call consumer.shutdown() twice, the second call throws Can not switch state while another state transition is in progress. error.

Actual:

     consumer.shutdown();
     try {
        consumer.shutdown();
      } catch (err) {
       // error caught as "Can not switch state while another state transition is in progress."
        console.error(err.message);
      }

Actual:

    consumer.shutdown();
    consumer.shutdown((err)=>{
        // unable to catch "Can not switch state while another state transition is in progress." error
    });

IMO, any error should be thrown to the callback in shutdown(cb).

Expect:

    consumer.shutdown();
    consumer.shutdown((err)=>{
        // catch error here
    });

Should the above line call this[sProcessMessageFailure] instead of throwing error?

Situation: Our extended consumer class's consume method has a promise, so if any error occurs inside that promise, its being unhandled by the consume method (even though there's a catch) and onConsumed throws that error and halts the process.

Please guide.

Hello everybody,
i had a problem when i run monitor service.

PS D:\server\chat-server> node .\monitor-queue.js
Monitor server is running...
D:\server\chat-server\node_modules\redis-smq\src\stats-frontend.js:204
if (err) throw err;
^
ReplyError: ERR wrong number of arguments for 'set' command
at parseError (D:\server\chat-server\node_modules\redis-parser\lib\parser.js:193:12)
at parseType (D:\server\chat-server\node_modules\redis-parser\lib\parser.js:303:14)

I am using node version v7.10.1.
Please help me!
Thank

Is there any plan to include a pause/ resume queue feature? May be on global level so no worker can consume from the queue until the queue is resumed.

Hi.

The monitoring not working in version 1.0.8, but his work in v1.0.7

TypeError: Cannot destructure property queueName of 'undefined' or 'null'.
at Object.getKeys (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\redis-keys.js:14:29)
at statsFrontend (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\stats-frontend.js:15:29)
at Server.server.listen (D:\nodeProjects\redissmqcollector\node_modules\redis-smq\src\monitor\index.js:52:31)
at Object.onceWrapper (events.js:313:30)
at emitNone (events.js:111:20)
at Server.emit (events.js:208:7)
at emitListeningNT (net.js:1378:10)
at _combinedTickCallback (internal/process/next_tick.js:135:11)
at process._tickCallback (internal/process/next_tick.js:180:9)
at Function.Module.runMain (module.js:678:11)

Thanks

We recently updated our redis-smq Node package from 1.0.22 to 1.1.1, and our consumers and producers started to fail because it would attempt to connect to 127.0.0.1 instead of the address specified in our config. Our config file looks like this:

module.exports = {
namespace: 'dpp-redis-server',
redis: {
host: 'our.host.here',
port: 6379
},
log: {
enabled: 1
},
monitor: {
enabled: true,
host: '127.0.0.1',
port: 3000,
},
};

We noticed that the example config file in the repo has a slightly different configuration than version 1.0.22. Once we updated the config file, then it works as expected.

However, the biggest issue is that npm updates from 1.0.22 to 1.1.1 without expecting breaking API changes.

Could you look into adding backwards compatibility between configs from version 1.0 into 1.1?