x.js

function createRedisStream(client, namespace, streamName, callback) {
    streamName = namespace + ":" + streamName
    var writableStream = client.stream("publish", streamName)
        , readableStream = client.stream("subscribe")
    
    readableStream.write(streamName)

    callback(es.duplex(writableStream, readableStream))
}