code: function processInput(buffer, cb) {
      if (buffer.length == 0) return buffer;
      var split_char = '\n';
      var lines = buffer.split(split_char);
      // If there are any line splits, the below logic always works, but if there are none we need to detect this and skip
      // any processing.
      if (lines[0].length == buffer.length) return buffer;
      // Note last item is ignored since it is the remainder (or empty)
      for(var i = 0; i < lines.length-1; i++) {
          var line = lines[i];
          if (format == "binary") {
              target.produce(line, handleProduceResponse.bind(undefined, cb));
          } else if (format == "avro") {
              // Avro data should be passed in its JSON-serialized form
              try {
                  var avro = JSON.parse(line);
              } catch (e) {
                  console.log("Couldn't parse '" + line + "' as JSON");
                  continue;
              }
              target.produce(valueSchema, avro, handleProduceResponse.bind(undefined, cb));
          }
          // OR with key or partition:
          //target.produce({'partition': 0, 'value': line}, handleProduceResponse.bind(undefined, cb));
          //target.produce({'key': 'console', 'value': line}, handleProduceResponse.bind(undefined, cb));
          num_messages += 1;
          num_bytes += line.length;
      }
      return lines[lines.length-1];
  }
docs: Splits input by lines into individual messages and passes them to the producer. Tracks stats to print at exit.