Merge pull request #19 from Kakulukian/master

[moleculer-bull] Add job concurrency
moleculerjs · Mar 18, 2018 · c89423c · c89423c
2 parents 9ae6a91 + d863637
commit c89423c
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 4 deletions.
diff --git a/packages/moleculer-bull/src/index.js b/packages/moleculer-bull/src/index.js
@@ -55,7 +55,10 @@ module.exports = function createService(url, queueOpts) {
 
 			if (this.schema.queues) {
 				_.forIn(this.schema.queues, (fn, name) => {
-					this.getQueue(name).process(fn.bind(this));
+					if(typeof fn === "function")
+						this.getQueue(name).process(fn.bind(this));
+					else
+						this.getQueue(name).process(fn.concurrency, fn.process.bind(this));
 				});
 			}
 

diff --git a/packages/moleculer-bull/test/unit/index.spec.js b/packages/moleculer-bull/test/unit/index.spec.js
@@ -37,20 +37,28 @@ describe("Test BullService created handler", () => {
 		queues: {
 			"task.first": jest.fn(),
 			"task.second": jest.fn(),
+			"task.concurrency": {
+				concurrency: 100,
+				process(job) {
+					return jest.fn();
+				},
+			},
 		}
 	});
 
 	it("should be created queues", () => {
 		expect(service).toBeDefined();
-		expect(Object.keys(service.$queues).length).toBe(2);
+		expect(Object.keys(service.$queues).length).toBe(3);
 		expect(service.$queues["task.first"]).toBeDefined();
 		expect(service.$queues["task.second"]).toBeDefined();
+		expect(service.$queues["task.concurrency"]).toBeDefined();
 
-		expect(Queue).toHaveBeenCalledTimes(2);
+		expect(Queue).toHaveBeenCalledTimes(3);
 		expect(Queue).toHaveBeenCalledWith("task.first", url, opts);
 		expect(Queue).toHaveBeenCalledWith("task.second", url, opts);
+		expect(Queue).toHaveBeenCalledWith("task.concurrency", url, opts);
 
-		expect(processCB).toHaveBeenCalledTimes(2);
+		expect(processCB).toHaveBeenCalledTimes(3);
 	});
 
 });