Aliyun Open Notification Service Client (base on opensource project RocketMQ)
Sub module of ali-sdk.
npm install ali-ons --save
consumer
'use strict';
const httpclient = require('urllib');
const Consumer = require('ali-ons').Consumer;
const consumer = new Consumer({
httpclient,
accessKeyId: 'your-accessKeyId',
accessKeySecret: 'your-AccessKeySecret',
consumerGroup: 'your-consumer-group',
// namespace: '', // aliyun namespace support
// isBroadcast: true,
});
consumer.subscribe(config.topic, '*', async msg => {
console.log(`receive message, msgId: ${msg.msgId}, body: ${msg.body.toString()}`)
// return Consumer.ACTION_RETRY; // you can return ACTION_RETRY, then this message will be directly retried
});
consumer.on('error', err => console.log(err));
If you want to use sql filter, you can subscribe a topic with a sql expression:
consumer.subscribe(
config.topic,
{
expressionType: 'SQL92',
subString: 'a is not null'
},
async msg => {
console.log(`receive message, msgId: ${msg.msgId}, body: ${msg.body.toString()}`)
}
);
For more information about sql filter, see: Filter Messages By SQL92
producer
'use strict';
const httpclient = require('urllib');
const Producer = require('ali-ons').Producer;
const Message = require('ali-ons').Message;
const producer = new Producer({
httpclient,
accessKeyId: 'your-accessKeyId',
accessKeySecret: 'your-AccessKeySecret',
producerGroup: 'your-producer-group',
// namespace: '', // aliyun namespace support
});
(async () => {
const msg = new Message('your-topic', // topic
'TagA', // tag
'Hello ONS !!! ' // body
);
// set Message#keys
msg.keys = ['key1'];
// delay consume
// msg.setStartDeliverTime(Date.now() + 5000);
const sendResult = await producer.send(msg);
console.log(sendResult);
})().catch(err => console.error(err))
Please contact to @gxcsoccer to give you accessKey