From 63b41da0ba9e6e14bd4b5f8c72992d72e156c596 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 12:06:18 +0200 Subject: wait for rabbit and db to start --- worker.js | 26 +++++++++++++++++++------- 1 file changed, 19 insertions(+), 7 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index c5c0582..a2287b9 100644 --- a/worker.js +++ b/worker.js @@ -5,7 +5,8 @@ var amqp = require("amqplib"), Seq = require("sequelize"), snakeCaseKeys = require("snakecase-keys"), - item_name_map = require("../orm/items"); + item_name_map = require("../orm/items"), + sleep = require("sleep-promise"); var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", @@ -13,10 +14,23 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { - let seq = new Seq(DATABASE_URI, { logging: () => {} }), - model = require("../orm/model")(seq, Seq), - rabbit = await amqp.connect(RABBITMQ_URI), - ch = await rabbit.createChannel(); + let seq, model, rabbit, ch; + + while (true) { + try { + seq = new Seq(DATABASE_URI, { logging: () => {} }), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + await ch.assertQueue("process", {durable: true}); + await ch.assertQueue("compile", {durable: true}); + break; + } catch (err) { + console.error(err); + await sleep(5000); + } + } + + model = require("../orm/model")(seq, Seq); let queue = [], timer = undefined; @@ -32,8 +46,6 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await model.Item.findAll() .map((item) => item_db_map[item.name] = item.id); - await ch.assertQueue("process", {durable: true}); - await ch.assertQueue("compile", {durable: true}); // as long as the queue is filled, msg are not ACKed // server sends as long as there are less than `prefetch` unACKed await ch.prefetch(BATCHSIZE); -- cgit v1.3.1