<commit_before>var redisUrl = process.env.MESSAGE_BUS_REDIS_URL || "127.0.0.1";
var redis = require("redis"),
    redisClient = redis.createClient(6379, redisUrl);
var AWS = require('aws-sdk');

AWS.config.region = 'us-east-1';
var kinesis = new AWS.Kinesis();

redisClient.on("ready", function () {
  console.log("Connected to Redis!");
  redisClient.psubscribe("*");
});

redisClient.on("pmessage", function (pattern, channel, message) {
  // var c = channel.split(":");
  // var model = c[0];
  // var action = c[1];
  // console.log("Model: ", model);
  // console.log("Action: ", action);
  // console.log("Message: ", message);

  var params = {
    Data: message, /* required */
    PartitionKey: channel, /* required */
    StreamName: 'mantle', /* required */
  };

  kinesis.putRecord(params, function(err, data) {
    if (err) {
      console.log('Error on channel: ' + channel);
      console.log(err, err.stack);
    }
    // else     console.log(data);           // successful response
  });
});

redisClient.on("error", function (err) {
  console.log("Redis error: " + err);
});

<commit_msg>Wrap up model and action in payload<commit_after>var redisUrl = process.env.MESSAGE_BUS_REDIS_URL || "127.0.0.1";
var redis = require("redis"),
    redisClient = redis.createClient(6379, redisUrl);
var AWS = require('aws-sdk');

AWS.config.region = 'us-east-1';
var kinesis = new AWS.Kinesis();

redisClient.on("ready", function () {
  console.log("Connected to Redis!");
  redisClient.psubscribe("*");
});

redisClient.on("pmessage", function (pattern, channel, message) {
  var c = channel.split(":");
  var model = c[0];
  var action = c[1];
  // console.log("Model: ", model);
  // console.log("Action: ", action);
  // console.log("Message: ", message);
  var msg = JSON.parse(message);
  var data = {
    model: model,
    action: action,
    message: msg
  }

  var params = {
    Data: JSON.stringify(data), /* required */
    PartitionKey: channel, /* required */
    StreamName: 'mantle', /* required */
  };

  kinesis.putRecord(params, function(err, data) {
    if (err) {
      console.log('Error on channel: ' + channel);
      console.log(err, err.stack);
    }
    // else     console.log(data);           // successful response
  });
});

redisClient.on("error", function (err) {
  console.log("Redis error: " + err);
});


{'fraction_non_alphanumeric': 0.18747337025990626, 'fraction_numerical': 0.011077971878994461, 'mean_word_length': 3.276867030965392, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7597771', 'n_tokens_mistral': 815, 'n_tokens_neox': 797, 'n_words': 171}