How to use with sockets?

This is question as opposed to an issue, I didn't see a mailing list, if there is a better place to post this, please let me know.

I am looking for a way to deal with socket streams, to control many connections with different parallelism per server, and to handle cases where requests need to be retried. Below is some coffeescript code that describes what I am looking for (client and server run in same process for testing). comments above "client" describe what I'm trying to do. Is highland the correct library to use for this task? If someone could convert the "client" side to the "highland way" It would really help with my understanding.

``` coffeescript
h = require "highland"
net = require "net"

HOST = "127.0.0.1"
createServer = (port) ->
net.createServer((sock) ->
#sock.write("welcome! on port #{port}\r\n")
#console.log("CONNECTED: #{sock.remoteAddress}:#{sock.remotePort}")
if port == 7001 # this is the work queue, what ports to send to...
setInterval (-> sock.write("700#{(Math.random() * 3 | 0) + 2}\r\n")), 0
return
sock.on "data", (data) ->
if (Math.random() * 3 | 0) + 1 == 3 # every 3rd
sock.write "wait 10 seconds and retry\n"
else
sock.write "#{data*2}\n"
sock.close

#sock.on "close", (data) ->
#console.log("CLOSED: #{sock.remoteAddress}:#{sock.remotePort}")
# return
return
).listen port, HOST
console.log "server listening on #{port}"
return

# create servers
for p in [7001..7004] # port 7001 is our work generator
createServer(p)

# CLIENT
# read from port 7001 to get a list of work to do.
# It sends out port numbers of where to send
# work. workers answer 1 question and then
# close socket.
#
# if worker sends back: "wait $x seconds", then
# wait that many seconds and retry with the work.
#
# would like to allow for 5 simultaneous connections to
# port 7002, 3 simultaneous connections to port 7003
# and 1 simultaneous connection to port 7004
#
# workers multiply number sent by 2. would like to
# keep a running total for each port and if total is
# mod 10000==0 then print it out with the port#
# keep a running total of answers per worker and
# print out number of answers every 1K (per port)

# here is code to connect to 7001 to get
# work and farm out to worker ports and track counts.
#
# how would I use highland to accomplish
# the things described above?
# this code is so messy...
# also this is somehow not closing sockets
# as after 20K per port I get:
# events.js:72
# throw er; // Unhandled 'error' event
# ^
#Error: connect EMFILE

g = net.createConnection(7001, HOST)
tots = {7002: 0, 7003: 0, 7004: 0 }
portcnt = {7002: 0, 7003: 0, 7004: 0 }

g.on "data", (data) ->
port = parseInt(data)
if isNaN(port) then return
w = net.createConnection(port, "127.0.0.1")
w.theport = port
w.write("50\n")
w.on "data", (d) ->
p = w.theport
ans = parseInt(d.toString())
if isNaN(ans)
w.end()
return
tots[p] += ans
portcnt[p] += 1
if (tots[p]%%10000)==0 then console.log "tots #{p}: #{tots[p]}"
if (portcnt[p]%%1000)==0 then console.log "portcnt #{p}: #{portcnt[p]}"
w.end()
return
return

```

Begin with the CLIENT section and the net.createConnection socket handlers shown in the issue. Compare the requested concurrency, retry, counting, and socket-closing behavior with Highland's stream API and Node.js socket documentation. Done would require a settled, documented approach for this use case, including the EMFILE failure.

Written by the indexing model from the issue text.