examples/console_producer.js
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];
}
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];
}
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 ] ; }
Splits input by lines into individual messages and passes them to the producer. Tracks stats to print at exit.
Splits input by lines into individual messages and passes them to the producer . Tracks stats to print at exit .