Nodes create duplex stream from readable and writable

Question

I work on adventures in school flows and get a duplex reduction challenge.

I solved the problem using the duplexer module, but wanted to solve it using my own stream functions.

Here is a solution using a duplexer:

var duplexer = require('duplexer');
var through = require('through');

module.exports = function (counter) {
    var counts = {};
    var input = through(write, end);
    return duplexer(input, counter);

    function write (row) {
        counts[row.country] = (counts[row.country] || 0) + 1;
    }
    function end () { counter.setCounts(counts) }
};

Thus, it receives a readable stream and a duplex stream returns. I am trying to solve this problem again:

var duplexer = require('duplexer');
var stream   = require('stream');

module.exports = function (counter) {
    var counts = {};
    var ts = stream.Transform({ objectMode: true })

    ts._transform = function(obj, enc, next) {
        var country = obj.country;
        var count = counts[country] || 0;
        counts[obj.country] = count + 1;
        next();
    }

    ts.on('finish', function() {
        counter.setCounts(counts);
    });

    counter.pipe(ts)
    return ts;
};

When started as is, this does not lead to a lack of output, so I changed next()to next(null, JSON.stringify(counts)). A string is because the next thread is not set to object mode. I get a conclusion, but its wrong, and it throws stream.push() after EOF.