Sonic Benchmark: Batch PUSH

sonic-benchmark_batch-push.js

var mongoose = require("mongoose");
var SonicChannelIngest = require("sonic-channel").Ingest;

var MessageModel = mongoose.model("message", new mongoose.Schema({
  session_id : String,
  website_id : String,
  type       : String,
  content    : Object
}));

var push_count = 0;
var last_count = 0;
var last_time_start = 0;

var worst_time = -1;
var best_time = -1;

var push_next = function(sonicChannelIngest, cursor) {
  cursor.next()
    .then((message) => {
      // Done?
      if (message === null) {
        console.info("All done. Bye.");

        process.exit(0);

        return;
      }

      var content = (message.content || "").trim();

      if (content) {
        var commited = sonicChannelIngest.push(
          "messages", message.website_id, message.session_id, content,

          (error) => {
            if (error) {
              console.error("Ingest error", error);
            }

            if (last_count++ === 1000) {
              last_count = 1;

              var date_now = Date.now();

              var time_taken = ((date_now - last_time_start) / 1000);

              if (worst_time === -1 || time_taken > worst_time) {
                worst_time = time_taken;
              }
              if (best_time === -1 || time_taken < best_time) {
                best_time = time_taken;
              }

              console.debug(
                "Step: cursor #" + push_count + "; " +
                  "took " + time_taken + " sec (worst " + worst_time + " sec;" +
                  " best " + best_time + " sec)"
              );

              last_time_start = date_now;
            }

            push_count++;

            push_next(sonicChannelIngest, cursor);
          }
        );

        if (commited !== true) {
          console.warn("Not commited yet: #" + last_count);
        }
      } else {
        push_next(sonicChannelIngest, cursor);
      }
    })
    .catch((error) => {
      console.error("Cursor error", error);

      process.exit(1);
    });
};

mongoose.connect("mongodb://localhost/messages", {
  useNewUrlParser : true
})
  .then(() => {
    console.info("Mongoose connected.");

    var sonicChannelIngest = new SonicChannelIngest({
      host : "::1",
      port : 1491,
      auth : "SecretPassword"
    }).connect({
      connected : function() {
        console.info("Sonic Channel succeeded to connect to host.");

        var cursor = MessageModel.find(
          {
            type : "text"
          },

          {
            website_id : true,
            session_id : true,
            content    : true
          }
        )
          .cursor();

        last_time_start = Date.now();

        push_next(sonicChannelIngest, cursor);
      },

      disconnected : function() {
        console.error("Sonic Channel is now disconnected.");

        process.exit(1);
      },

      timeout : function() {
        console.error("Sonic Channel connection timed out.");
      },

      retrying : function() {
        console.error("Trying to reconnect to Sonic Channel...");
      },

      error : function(error) {
        console.error("Sonic Channel failed to connect to host.", error);
      }
    });
  })
  .catch((error) => {
    console.error("Mongoose connect failed.", error);
  });