message: Update deli to restart on mongo errors
diff: import { queue } from "async";
  import * as kafka from "kafka-node";
  import * as _ from "lodash";
  -import { Collection, MongoClient } from "mongodb";
  +import { Collection } from "mongodb";
  import * as nconf from "nconf";
  import * as path from "path";
  import * as core from "../core";
  @@ -55,28 +55,26 @@ function processMessage(
  async function checkpoint(
  partitionManager: core.PartitionManager,
  - dispensers: { [key: string]: TakeANumber },
  + dispensers: TakeANumber[],
  pendingTickets: Array<Promise<void>>) {
  // Ticket all messages and empty the queue.
  await Promise.all(pendingTickets);
  // Checkpoint to mongo now.
  - let checkpointQueue = [];
  - for (let doc of Object.keys(dispensers)) {
  - checkpointQueue.push(dispensers[doc].checkpoint());
  - }
  + let checkpointQueue = dispensers.map((dispenser) => dispenser.checkpoint());
  await Promise.all(checkpointQueue);
  // Finally call kafka checkpointing.
  partitionManager.checkPoint();
  }
  -async function processMessages(
  +function processMessages(
  kafkaClient: kafka.Client,
  producer: utils.kafka.Producer,
  - objectsCollection: Collection) {
  + objectsCollection: Collection): Promise<void> {
  + const deferred = new utils.Deferred<void>();
  const dispensers: { [key: string]: TakeANumber } = {};
  const consumerOffset = new kafka.Offset(kafkaClient);
  @@ -101,7 +99,7 @@ async function processMessages(
  // https://github.com/SOHU-Co/kafka-node/issues/90
  console.error(`Error in kafka consumer: ${error}. Wait for 30 seconds and restart...`);
  setTimeout(() => {
  - process.exit(1);
  + deferred.reject(error);
  }, 30000);
  });
  @@ -117,10 +115,11 @@ async function processMessages(
  // Periodically checkpoints to mongo and checkpoints offset back to kafka.
  // Ideally there should be a better strategy to figure out when to checkpoint.
  if (message.offset % checkpointBatchSize === 0) {
  + const pendingDispensers = _.keys(ticketQueue).map((key) => dispensers[key]);
  const pendingTickets = _.values(ticketQueue);
  ticketQueue = {};
  - checkpoint(partitionManager, dispensers, pendingTickets).catch((error) => {
  - console.error(error);
  + checkpoint(partitionManager, pendingDispensers, pendingTickets).catch((error) => {
  + deferred.reject(error);
  });
  }
  @@ -131,12 +130,14 @@ async function processMessages(
  throughput.produce();
  q.push(message);
  });
  +
  + return deferred.promise;
  }
  async function run() {
  // Connection to stored document details
  - const client = await MongoClient.connect(mongoUrl);
  - console.log("Connected to Mongo");
  + const mongoManager = new utils.MongoManager(mongoUrl, false);
  + const client = await mongoManager.getDatabase();
  const objectsCollection = await client.collection(objectsCollectionName);
  console.log("Collection ready");
  