Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

use promisified smqp as backend #5

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .nvmrc
Original file line number Diff line number Diff line change
@@ -1 +1 @@
8
10
2 changes: 1 addition & 1 deletion .travis.yml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
language: node_js
node_js:
- 8
- 9
- 10
- 12
sudo: false
186 changes: 99 additions & 87 deletions index.js
Original file line number Diff line number Diff line change
@@ -1,104 +1,116 @@
"use strict";

let exchanges = {};
let queues = {};
const {Broker} = require("smqp");

module.exports = {connect, resetMock};
const connections = [];

function connect(url, options, connCallback) {
if (!connCallback) {
options = {};
connCallback = options;
}
module.exports = Fake();

const connection = {
createChannel: createChannel(false),
createConfirmChannel: createChannel(true),
on: function () {},
close: resetMock,
function Fake() {
return {
connections,
resetMock,
connect,
};

return connCallback(null, connection);

function createChannel(confirm) {
return (channelCallback) => {
channelCallback(null, {
assertQueue,
assertExchange,
bindQueue,
publish,
consume,
deleteQueue,
ack,
nack,
prefetch,
on,
});

function assertQueue(queue, qOptions, qCallback) {
qCallback = qCallback || function () {};
setIfUndef(queues, queue, {messages: [], subscribers: [], options: qOptions});
qCallback();
}
function connect(amqpUrl, ...args) {
const {_broker} = connections.find((conn) => conn.options[0] === amqpUrl) || {};
const broker = _broker || Broker();
const connection = Connection(broker, amqpUrl, ...args);
connections.push(connection);
return resolveOrCallback(args.slice(-1)[0], null, connection);
}

function assertExchange(exchange, type, exchOptions, exchCallback) {
if (typeof (exchOptions) === "function") {
exchCallback = exchOptions;
exchOptions = {};
}
setIfUndef(exchanges, exchange, {bindings: [], options: exchOptions, type: type});
return exchCallback && exchCallback();
}
function resetMock() {
for (const connection of connections.slice()) {
connection.close();
}
}

function bindQueue(queue, exchange, key, args, bindCallback) {
bindCallback = bindCallback || function () {};
if (!exchanges[exchange]) return bindCallback("Bind to non-existing exchange " + exchange);
const re = "^" + key.replace(".", "\\.").replace("#", "(\\S)+").replace("*", "\\w+") + "$";
exchanges[exchange].bindings.push({regex: new RegExp(re), queueName: queue});
bindCallback();
}
function Connection(broker, ...connArgs) {
const options = connArgs.filter((a) => typeof a !== "function");

function publish(exchange, routingKey, content, props, pubCallback) {
pubCallback = pubCallback || function () {};
if (!exchanges[exchange]) return pubCallback("Publish to non-existing exchange " + exchange);
const bindings = exchanges[exchange].bindings;
const matchingBindings = bindings.filter((b) => b.regex.test(routingKey));
matchingBindings.forEach((binding) => {
const subscribers = queues[binding.queueName] ? queues[binding.queueName].subscribers : [];
subscribers.forEach((sub) => {
const message = {fields: {routingKey: routingKey}, properties: props, content: content};
sub(message);
});
});
if (confirm) pubCallback();
return true;
}
return {
_broker: broker,
options,
createChannel(...args) {
return resolveOrCallback(args.slice(-1)[0], null, Channel(broker));
},
createConfirmChannel(...args) {
return resolveOrCallback(args.slice(-1)[0], null, Channel(broker, true));
},
close(...args) {
const idx = connections.indexOf(this);
if (idx > -1) connections.splice(idx, 1);
broker.reset();
return resolveOrCallback(args.slice(-1)[0]);
},
on() {},
};
}

function consume(queue, handler) {
queues[queue].subscribers.push(handler);
}
function Channel(broker, confirm) {
return {
assertExchange(...args) {
return callBroker(broker.assertExchange, ...args);
},
assertQueue(...args) {
return callBroker(broker.assertQueue, ...args);
},
bindQueue(...args) {
return callBroker(broker.bindQueue, ...args);
},
deleteQueue(...args) {
return callBroker(broker.deleteQueue, ...args);
},
publish(exchange, routingKey, content, ...args) {
if (!Buffer.isBuffer(content)) throw new TypeError("content is not a buffer");
return confirm ? callBroker(broker.publish, exchange, routingKey, content, ...args) : broker.publish(exchange, routingKey, content, ...args);
},
sendToQueue(queue, content, ...args) {
if (!Buffer.isBuffer(content)) throw new TypeError("content is not a buffer");
return confirm ? callBroker(broker.sendToQueue, queue, content, ...args) : broker.sendToQueue(queue, content, ...args);
},
consume(queue, onMessage, ...args) {
const passArgs = args.length ? args : [{}];
return callBroker(broker.consume, queue, onMessage && handler, ...passArgs);
function handler(_, msg) {
onMessage(msg);
}
},
cancel(...args) {
return callBroker(broker.cancel, ...args);
},
ack: broker.ack,
ackAll: broker.ackAll,
nack: broker.nack,
reject: broker.reject,
nackAll: broker.nackAll,
prefetch() {},
on: broker.on,
};

function deleteQueue(queue) {
setImmediate(() => {
delete queues[queue];
});
}
function callBroker(fn, ...args) {
let [poppedCb] = args.slice(-1);
if (typeof poppedCb === "function") args.splice(-1);
else poppedCb = null;

function ack() {}
function nack() {}
function prefetch() {}
function on() {}
};
return new Promise((resolve, reject) => {
try {
const result = fn(...args);
if (poppedCb) poppedCb(null, result);
return resolve(result);
} catch (err) {
if (!poppedCb) return reject(err);
poppedCb(err);
return resolve();
}
});
}
}
}

function resetMock() {
queues = {};
exchanges = {};
}

function setIfUndef(object, prop, value) {
if (!object[prop]) {
object[prop] = value;
}
function resolveOrCallback(optionalCb, err, ...args) {
if (typeof optionalCb === "function") optionalCb(err, ...args);
return Promise.resolve(...args);
}
1,368 changes: 914 additions & 454 deletions package-lock.json

Large diffs are not rendered by default.

14 changes: 9 additions & 5 deletions package.json
Original file line number Diff line number Diff line change
@@ -7,7 +7,8 @@
"Mattias Norlander"
],
"scripts": {
"test": "mocha --reporter spec && eslint . --cache"
"test": "mocha -R spec -t 1000",
"posttest": "eslint . --cache"
},
"main": "index.js",
"engines": {
@@ -26,12 +27,15 @@
"url": "http://github.com/ExpressenAB/exp-fake-amqplib/raw/master/LICENSE"
},
"devDependencies": {
"chai": "^4.1.2",
"eslint": "^5.4.0",
"mocha": "^5.2.0"
"chai": "^4.2.0",
"eslint": "^6.7.2",
"mocha": "^6.2.2"
},
"files": [
"index.js",
"LICENSE"
]
],
"dependencies": {
"smqp": "^2.1.1"
}
}
219 changes: 217 additions & 2 deletions test/index.js
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
"use strict";

const {connect} = require("..");
const {connect, resetMock, connections} = require("..");
const {expect} = require("chai");

describe("fake amqplib", () => {
describe(".connect()", () => {
describe(".connect([amqpUrl])", () => {
it("exposes the expected api on connection", (done) => {
connect("amqp://localhost", null, (err, connection) => {
if (err) return done(err);
@@ -15,6 +15,33 @@ describe("fake amqplib", () => {
done();
});
});

it("connection with the same amqpUrl shares broker", async () => {
const conn1 = await connect("amqp://testrabbit:5672");
const conn2 = await connect("amqp://testrabbit:5672");

expect(conn1._broker === conn2._broker).to.be.true;
});

it("connection with different amqpUrls has different brokers", async () => {
const conn1 = await connect("amqp://testrabbit:5672");
const conn2 = await connect("amqp://testrabbit:15672");

expect(conn1._broker === conn2._broker).to.be.false;
});

it("exposes connection list", async () => {
const conn1 = await connect("amqp://localhost:5672");
const conn2 = await connect("amqp://localhost:15672");
expect(connections).to.have.length.above(2).and.include.members([conn1, conn2]);
});

it("connection.close() removes connection from list", async () => {
const conn = await connect("amqp://testrabbit:5672");
expect(connections).to.include(conn);

conn.close();
});
});

describe("channels", () => {
@@ -60,5 +87,193 @@ describe("fake amqplib", () => {
done();
});
});

it("createChannel returns a promise with resolved channel", async () => {
const channel = await connection.createChannel();
expect(channel).to.have.property("assertExchange").that.is.a("function");
});

it("can assert exchange into existance", (done) => {
connection.createChannel((err, channel) => {
if (err) return done(err);
channel.assertExchange("event", "topic", () => {
done();
});
});
});

it("returns error in callback", (done) => {
connection.createChannel((channelErr, channel) => {
if (channelErr) return done(channelErr);
channel.assertExchange("wrong-type", {}, (err) => {
expect(err).to.be.ok.and.have.property("message").that.match(/topic or direct/);
done();
});
});
});

it("returns promise that can be caught", async () => {
const channel = await connection.createChannel();
const err = await channel.assertExchange("event", "directly").catch((e) => e);
expect(err).to.be.ok.and.have.property("message");
});

it("throws if unsupported function is called", async () => {
const channel = await connection.createChannel();
expect(() => {
channel.subscribeOnce("event");
}).to.throw(Error, /is not a function/);
});

it("throws if consume() is called without message callback", async () => {
const channel = await connection.createChannel();
await channel.assertQueue("event-q");
try {
await channel.consume("event-q");
} catch (e) {
var err = e; // eslint-disable-line
}
expect(err).to.be.ok;
expect(err.message).to.match(/Message callback/i);
});

it("consume() returns promise", async () => {
const channel = await connection.createChannel();
await channel.assertQueue("event-q");
const ok = await channel.consume("event-q", onMessage).then((result) => result);

expect(ok).to.be.ok.and.have.property("consumerTag").that.is.ok;

function onMessage() {}
});

it("consume() returns message with excpected arguments in message callback", async () => {
const channel = await connection.createChannel();
await channel.assertExchange("consume");
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");
await channel.consume("consume-q", onMessage).then((result) => result);

let onMessageArgs;

await channel.publish("consume", "test", Buffer.from(JSON.stringify({data: 1})));

expect(onMessageArgs, "message arguments").to.be.ok;
expect(onMessageArgs.length).to.equal(1);
const msg = onMessageArgs[0];

expect(msg).to.have.property("fields").with.property("routingKey", "test");

function onMessage(...args) {
onMessageArgs = args;
}
});
});

describe("publish", () => {
let connection;
before((done) => {
connect("amqp://localhost", null, (err, conn) => {
if (err) return done(err);
connection = conn;
done();
});
});

it("ignores callback", async () => {
const channel = await connection.createChannel();
await channel.assertExchange("consume");
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");

return new Promise((resolve, reject) => {
channel.publish("consume", "test.1", Buffer.from("msg"), {type: "test"}, () => {
reject(new Error("Ignore callback"));
});
channel.consume("consume-q", resolve);
});
});

it("confirm channel calls callback when published", async () => {
const channel = await connection.createConfirmChannel();
await channel.assertExchange("consume");
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");

return new Promise((resolve) => {
channel.publish("consume", "test.1", Buffer.from(JSON.stringify({})), () => {
resolve();
});
});
});
});

describe("sendToQueue", () => {
let connection;
before((done) => {
connect("amqp://localhost", null, (err, conn) => {
if (err) return done(err);
connection = conn;
done();
});
});

it("breaks if message is not a buffer", async () => {
const channel = await connection.createChannel();
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");

expect(() => channel.sendToQueue("consume-q", {})).to.throw(/not a buffer/i);
});

it("ignores callback", async () => {
const channel = await connection.createChannel();
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");

return new Promise((resolve, reject) => {
channel.sendToQueue("consume-q", Buffer.from("msg"), () => {
reject(new Error("Ignore callback"));
});
channel.consume("consume-q", resolve);
});
});

it("confirm channel calls callback when sent", async () => {
const channel = await connection.createConfirmChannel();
await channel.assertQueue("consume-q");
await channel.bindQueue("consume-q", "consume", "#");

return new Promise((resolve) => {
channel.sendToQueue("consume-q", Buffer.from(JSON.stringify({})), () => {
resolve();
});
});
});
});

describe("resetMock", () => {
it("clears queues, exchanges, and consumers", async () => {
const connection = await connect("amqp://localhost:15672");
const channel = await connection.createChannel();
await channel.assertExchange("event", "topic", {durable: true, autoDelete: false});
await channel.assertQueue("event-q");

await channel.bindQueue("event-q", "event", "#", {durable: true});

await channel.assertExchange("temp", "topic", {durable: false});
await channel.assertQueue("frifras-q", {durable: false});

await channel.bindQueue("frifras-q", "temp", "#");

await channel.publish("event", "test", Buffer.from("msg1"));
await channel.publish("temp", "test", Buffer.from("msg2"));

resetMock();

expect(connection._broker).to.have.property("exchangeCount", 0);
expect(connection._broker).to.have.property("queueCount", 0);
expect(connection._broker).to.have.property("consumerCount", 0);
});
});
});