8000 refactored Work Queues tutorial · zerolugithub/rabbitmq-tutorials@b818682 · GitHub
[go: up one dir, main page]
Skip to content

Commit b818682

Browse files
priviteragpriviterag
committed
refactored Work Queues tutorial
1 parent 2030f47 commit b818682

File tree

2 files changed

+24
-49
lines changed

2 files changed

+24
-49
lines changed

‎javascript-nodejs/src/new_task.js

Lines changed: 8 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,15 @@
11
#!/usr/bin/env node
22

3-
var amqp = require('amqplib');
4-
var when = require('when');
3+
var amqp = require('amqplib/callback_api');
54

6-
var conn = amqp.connect('amqp://localhost');
7-
conn.then(createChannel).then(null, console.warn);
8-
9-
function createChannel(conn) {
10-
return when(
11-
conn.createChannel().
12-
then(sendMessage)).
13-
ensure(function() {
14-
conn.close();
15-
});
16-
}
17-
18-
function sendMessage(ch) {
19-
var q = 'task_queue';
20-
var ok = ch.assertQueue(q, {durable: true});
21-
22-
return ok.then(function(_ignore) {
5+
amqp.connect('amqp://localhost', function(err, conn) {
6+
conn.createChannel(function(err, ch) {
7+
var q = 'task_queue';
238
var msg = process.argv.slice(2).join(' ') || "Hello World!";
9+
10+
ch.assertQueue(q, {durable: true});
2411
ch.sendToQueue(q, new Buffer(msg), {persistent: true});
2512
console.log(" [x] Sent '%s'", msg);
26-
return ch.close();
2713
});
28-
}
14+
setTimeout(function() { conn.close(); process.exit(0) }, 500);
15+
});

‎javascript-nodejs/src/worker.js

Lines changed: 16 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,22 @@
11
#!/usr/bin/env node
22

3-
var amqp = require('amqplib');
3+
var amqp = require('amqplib/callback_api');
44

5-
var conn = amqp.connect('amqp://localhost');
6-
conn.then(createChannel).then(null, console.warn);
5+
amqp.connect('amqp://localhost', function(err, conn) {
6+
conn.createChannel(function(err, ch) {
7+
var q = 'task_queue';
78

8-
function createChannel(conn) {
9-
process.once('SIGINT', function() { conn.close(); });
10-
return conn.createChannel().then(consume);
11-
}
9+
ch.assertQueue(q, {durable: true});
10+
ch.prefetch(1);
11+
console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q);
12+
ch.consume(q, function(msg) {
13+
var secs = msg.content.toString().split('.').length - 1;
1214

13-
function consume(ch) {
14-
var ok = ch.assertQueue('task_queue', {durable: true});
15-
16-
ok = ok.then(function() { ch.prefetch(1); });
17-
ok = ok.then(function(_ignore) {
18-
return ch.consume(
19-
'task_queue',
20-
function doSomeWork(msg) {
21-
console.log(" [x] Received '%s'", msg.content.toString());
22-
var secs = msg.content.toString().split('.').length - 1;
23-
setTimeout(function() {
24-
console.log(" [x] Done");
25-
ch.ack(msg);
26-
}, secs * 1000);
27-
},
28-
{noAck: false});
29-
});
30-
31-
return ok.then(function(_consumeOk) {
32-
console.log(' [*] Waiting for messages. To exit press CTRL+C');
15+
console.log(" [x] Received %s", msg.content.toString());
16+
setTimeout(function() {
17+
console.log(" [x] Done");
18+
ch.ack(msg);
19+
}, secs * 1000);
20+
}, {noAck: false});
3321
});
34-
}
22+
});

0 commit comments

Comments
 (0)
0