Skip to content

Commit 9409512

Browse files
BDav24BDav24
committed
remove use of sdk internal queue logic
1 parent 65c6e97 commit 9409512

11 files changed

Lines changed: 276 additions & 201 deletions

File tree

‎README.md‎

Lines changed: 53 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -34,89 +34,80 @@ $ yarn add notifme-sdk notifme-sdk-queue-rabbitmq
3434
```
3535

3636
```javascript
37-
import NotifmeSdk from 'notifme-sdk'
38-
import rabbitMq from 'notifme-sdk-queue-rabbitmq'
37+
// In your application
38+
import {NotifmeRabbitMqProducer} from 'notifme-sdk-queue-rabbitmq'
3939

40-
const notifmeSdk = new NotifmeSdk({
41-
requestQueue: rabbitMq('amqp://localhost'),
42-
runWorker: true, // consumer will start on the same instance than producer
43-
channels: {
44-
sms: {
45-
providers: [{type: 'logger'}]
46-
}
47-
},
48-
onError: function (result, request) {
49-
if (result.errors && result.errors.queue) {
50-
/*
51-
* Queue system is down! Decide what to do in this case. You can either:
52-
* - Send notifications with this instance (code below)
53-
* - Use an alternative queue system (and run a worker corresponding to the queue)
54-
* - (To simply retry later, pass `keepRequestsInMemoryWhileConnecting: true` in the options)
55-
*/
56-
notifmeSdk.sender.send(request) // providers need to be defined in NotifmeSdk options
57-
}
58-
// Also handle provider errors...
59-
}
40+
const notificationService = new NotifmeRabbitMqProducer({
41+
url: 'amqp://localhost'
6042
})
6143

62-
notifmeSdk
63-
.send({sms: {from: '+15000000000', to: '+15000000001', text: 'Hello, how are you?'}})
64-
.then(console.log)
44+
notificationService.enqueueNotification({
45+
sms: {from: '+15000000000', to: '+15000000001', text: 'Hello, how are you?'}
46+
})
6547
```
6648

67-
## How to use
49+
```javascript
50+
// In your worker
51+
import NotifmeSdk from 'notifme-sdk'
52+
import {NotifmeRabbitMqConsumer} from 'notifme-sdk-queue-rabbitmq'
6853

69-
### Options
54+
const notifmeSdk = new NotifmeSdk({
55+
// Define all your providers here.
56+
// (see documentation: https://github.com/notifme/notifme-sdk#2-providers)
57+
})
7058

71-
```javascript
72-
new NotifmeSdk({
73-
requestQueue: rabbitMq('URL (Example: amqp://localhost)', {
74-
amqpOptions: ...,
75-
reconnectDelaySecond: ...,
76-
keepRequestsInMemoryWhileConnecting: ...
77-
}),
78-
...
59+
const notifmeWorker = new NotifmeRabbitMqConsumer(notifmeSdk, {
60+
url: 'amqp://localhost'
7961
})
80-
```
62+
notifmeWorker.run()
8163

82-
| Option name | Required | Type | Description |
83-
| --- | --- | --- | --- |
84-
| `amqpOptions` | `false` | `Object` | Connection options. See [amqplib documentation](http://www.squaremobius.net/amqp.node/channel_api.html#connect). |
85-
| `reconnectDelaySecond` | `false` | `number` | <i>Default: `30`.</i><br>Time in second to wait between two reconnection tries. |
86-
| `keepRequestsInMemoryWhileConnecting` | `false` | `boolean` | <i>Default: `false`.</i><br>Should the requests be kept in memory while queue is (re)connecting? If set to `true`, may cause memory overflow. |
64+
```
8765

88-
See also [Notif.me documentation](https://github.com/notifme/notifme-sdk).
66+
See a [complete working example](https://github.com/notifme/notifme-sdk-queue-rabbitmq/tree/master/example) for more details.
8967

90-
### Use workers to send notifications
68+
## How to use
9169

92-
Use `runWorker: false` in the configuration of Notif.me SDK.
70+
### Producer options
9371

9472
```javascript
95-
// Producer
96-
const notifmeSdk = new NotifmeSdk({
97-
requestQueue: rabbitMq('amqp://localhost'),
98-
runWorker: false
73+
new NotifmeRabbitMqProducer({
74+
keepRequestsInMemoryWhileConnecting: ...,
75+
url: ...,
76+
amqpOptions: ...,
77+
queueName: ...,
78+
isPersistent: ...,
79+
reconnectDelaySecond: ...
9980
})
100-
101-
notifmeSdk.send(...)
10281
```
10382

83+
| Option name | Type | Default | Description |
84+
| --- | --- | --- | --- |
85+
| `keepRequestsInMemoryWhileConnecting` | `boolean` | `false` | Should the requests be kept in memory while queue is (re)connecting? If set to `true`, may cause memory overflow. |
86+
| `url` | `string` | `'amqp://localhost'` | RabbitMQ URL. See [amqplib documentation](http://www.squaremobius.net/amqp.node/channel_api.html#connect). |
87+
| `amqpOptions` | `Object` | `{}` | Connection options. See [amqplib documentation](http://www.squaremobius.net/amqp.node/channel_api.html#connect). |
88+
| `queueName` | `string` | `'notifme:request'` | Name of the queue to use. |
89+
| `isPersistent` | `boolean` | `true` | Is the queue persistent? |
90+
| `reconnectDelaySecond` | `number` | `30` | Time in second to wait between two reconnection tries. |
91+
92+
### Consumer options
93+
10494
```javascript
105-
// Consumer
106-
import NotifmeWorker from 'notifme-sdk/lib/worker'
107-
108-
const notifmeWorker = new NotifmeWorker({
109-
requestQueue: rabbitMq('amqp://localhost'),
110-
channels: {
111-
email: {
112-
providers: [{type: 'logger'}]
113-
}
114-
}
95+
new NotifmeRabbitMqConsumer({
96+
url: ...,
97+
amqpOptions: ...,
98+
queueName: ...,
99+
isPersistent: ...,
100+
reconnectDelaySecond: ...
115101
})
116-
notifmeWorker.run()
117102
```
118103

119-
See a [working example](https://github.com/notifme/notifme-sdk-queue-rabbitmq/tree/master/example) for more details.
104+
| Option name | Type | Default | Description |
105+
| --- | --- | --- | --- |
106+
| `url` | `string` | `'amqp://localhost'` | RabbitMQ URL. See [amqplib documentation](http://www.squaremobius.net/amqp.node/channel_api.html#connect). |
107+
| `amqpOptions` | `Object` | `{}` | Connection options. See [amqplib documentation](http://www.squaremobius.net/amqp.node/channel_api.html#connect). |
108+
| `queueName` | `string` | `'notifme:request'` | Name of the queue to use. |
109+
| `isPersistent` | `boolean` | `true` | Is the queue persistent? |
110+
| `reconnectDelaySecond` | `number` | `30` | Time in second to wait between two reconnection tries. |
120111

121112
## Contributing
122113

‎example/application.js‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/* @flow */
2+
import {NotifmeRabbitMqProducer} from '../src' // notifme-sdk-queue-rabbitmq
3+
4+
const notificationService = new NotifmeRabbitMqProducer({
5+
url: 'amqp://localhost'
6+
})
7+
8+
const notificationRequest = {
9+
email: {
10+
from: 'me@example.com',
11+
to: 'john@example.com',
12+
subject: 'Hi John',
13+
html: '<b>Hello John! How are you?</b>'
14+
}
15+
}
16+
17+
async function sendTestEmail () {
18+
try {
19+
await notificationService.enqueueNotification(notificationRequest)
20+
console.log('queued')
21+
} catch (error) {
22+
/*
23+
* Queue system is down!
24+
*
25+
* Decide what to do in this case (while queue tries to reconnect). You can either:
26+
* 1. Send notifications with this instance:
27+
* new notifmeSdk({your providers' options...}).send(notificationRequest)
28+
* 2. Use an alternative queue system (and run a worker corresponding to the queue)
29+
* 3. (To simply keep request in memory and retry later, pass `keepRequestsInMemoryWhileConnecting: true`
30+
* in the NotifmeRabbitMqProducer options and no error will be raised)
31+
*/
32+
console.error(error)
33+
}
34+
}
35+
36+
// Enqueue the notifications
37+
setInterval(sendTestEmail, 2000)

‎example/consumer.js‎

Lines changed: 0 additions & 15 deletions
This file was deleted.

‎example/index.js‎

Lines changed: 0 additions & 3 deletions
This file was deleted.

‎example/producer.js‎

Lines changed: 0 additions & 39 deletions
This file was deleted.

‎example/worker.js‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
/* @flow */
2+
import NotifmeSdk from 'notifme-sdk'
3+
import {NotifmeRabbitMqConsumer} from '../src' // notifme-sdk-queue-rabbitmq
4+
5+
const notifmeSdk = new NotifmeSdk({
6+
channels: {
7+
/*
8+
* Define all your providers here.
9+
* (see documentation: https://github.com/notifme/notifme-sdk#2-providers)
10+
*/
11+
email: {
12+
providers: [{type: 'logger'}]
13+
}
14+
}
15+
})
16+
17+
const notifmeWorker = new NotifmeRabbitMqConsumer(notifmeSdk, {
18+
url: 'amqp://localhost'
19+
})
20+
notifmeWorker.run()

‎package.json‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212
"clean": "rm -rf lib/",
1313
"dev": "nodemon -e js -r babel-register example/index.js",
1414
"lint": "yarn run flow && standard",
15+
"run-consumer": "babel-node example/worker.js",
16+
"run-producer": "babel-node example/application.js",
1517
"run-rabbitmq": "docker run -it --rm -p 5672:5672 rabbitmq:alpine",
1618
"prepublish": "yarn run build",
1719
"test": "yarn run lint"

‎src/consumer.js‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/* @flow */
2+
import RabbitMqQueue from './queue'
3+
import NotifmeSdk from 'notifme-sdk'
4+
// Types
5+
import type {NotificationRequestType} from 'notifme-sdk'
6+
import type {QueueOptionsType} from './queue'
7+
8+
export type ConsumerOptionsType = QueueOptionsType
9+
10+
export default class NotifmeRabbitMqConsumer {
11+
queue: RabbitMqQueue
12+
sender: NotifmeSdk
13+
14+
constructor (notifmeSdk: NotifmeSdk, options?: ConsumerOptionsType = {}) {
15+
this.queue = new RabbitMqQueue(options)
16+
this.sender = notifmeSdk
17+
}
18+
19+
run () {
20+
this.queue.registerConsumer(async (request: NotificationRequestType) => {
21+
await this.sender.send(request)
22+
})
23+
}
24+
}

‎src/index.js‎

Lines changed: 5 additions & 82 deletions
Original file line numberDiff line numberDiff line change
@@ -1,83 +1,6 @@
1-
const amqplib = require('amqplib')
1+
/* @flow */
2+
import Consumer from './consumer'
3+
import Producer from './producer'
24

3-
const defaultOptions = {
4-
amqpOptions: {},
5-
reconnectDelaySecond: 30,
6-
keepRequestsInMemoryWhileConnecting: false // /!\ if true, may cause memory overflow
7-
}
8-
9-
function NotifmeRabbitMq (url, options) {
10-
const mergedOptions = Object.assign({}, defaultOptions, options)
11-
const reconnectDelay = mergedOptions.reconnectDelaySecond
12-
13-
let connecting = true
14-
15-
function connect () {
16-
return amqplib.connect(url, mergedOptions.amqpOptions)
17-
.then(function (connection) {
18-
connection.on('error', reconnect)
19-
connection.on('close', reconnect)
20-
21-
console.log('[RabbitMQ] Connection successful.')
22-
connecting = false
23-
return connection.createChannel()
24-
})
25-
.catch(function () {
26-
return new Promise(function (resolve) {
27-
console.error(`[RabbitMQ] Connection error. Retry in ${reconnectDelay} seconds.`)
28-
setTimeout(function () { resolve(connect()) }, reconnectDelay * 1000)
29-
})
30-
})
31-
}
32-
33-
function reconnect () {
34-
connecting = true
35-
channelPromise = connect()
36-
channelPromise.then(function (channel) {
37-
Object.keys(consumers).forEach(function (type) {
38-
consumers[type].forEach(function (consumer) {
39-
addConsumer(type, consumer)
40-
})
41-
})
42-
})
43-
}
44-
45-
let channelPromise = connect()
46-
let consumers = {}
47-
48-
function addConsumer (type, callback) {
49-
return channelPromise.then(function (channel) {
50-
return channel.assertQueue(type, {durable: true}).then(function () {
51-
channel.prefetch(1)
52-
return channel.consume(type, function (message) {
53-
if (message !== null) {
54-
callback(JSON.parse(message.content.toString())).then(function () {
55-
channel.ack(message)
56-
})
57-
}
58-
}, {noAck: false})
59-
})
60-
})
61-
}
62-
63-
return {
64-
enqueue: (type, request) => {
65-
return (connecting && !mergedOptions.keepRequestsInMemoryWhileConnecting)
66-
? Promise.reject(new Error('[RabbitMQ] Queue is not ready yet. Please retry later.'))
67-
: channelPromise.then(function (channel) {
68-
return channel.assertQueue(type, {durable: true}).then(function () {
69-
return channel.sendToQueue(type, Buffer.from(JSON.stringify(request)), {persistent: true})
70-
})
71-
})
72-
},
73-
dequeue: (type, callback) => {
74-
if (!consumers[type]) {
75-
consumers[type] = []
76-
}
77-
consumers[type].push(callback)
78-
return addConsumer(type, callback)
79-
}
80-
}
81-
}
82-
83-
exports.default = NotifmeRabbitMq
5+
export const NotifmeRabbitMqConsumer = Consumer
6+
export const NotifmeRabbitMqProducer = Producer

0 commit comments

Comments
 (0)