socket.io readable stream

client.coffee

socket.current = null

socket.on 'start', (query) ->
  socket.current = query.name
  ($ '#main').html ' '
  console.log 'start'

socket.on 'data', (query, data) ->
  if socket.current is query.name
    ($ '#main').append (_.template ($ '#template').html()) data

socket.on 'end', (query) ->
  console.log 'end'

query = 'filename'        # とか
query = { name: /hoge/i } # とか

socket.emit 'fetch', query

server.class.coffee

Stream = require 'stream'

class WritableSocketStream extends Stream.Writable
  constructor: (@socket, options = {}) ->
    @socket.emit 'start'

  write: (chunk, encoding, callback) ->
    @socket.emit 'data', chunk.toString(encoding)

  end: (chunk, encoding, callback) ->
    @socket.emit 'data', chunk.toString(encoding) if chunk
    @socket.emit 'end'

io.sockets.on 'connection', (socket) ->
  socket.on 'fetch', (query) ->
    stream = fs.createReadStream path.resolve 'public', query
    stream.pipe new WritableSocketStream socket

    # あるいは

    stream = Data.find(query).stream()
    stream.pipe new WritableSocketStream socket

server.coffee

socket.current = null

socket.on 'fetch', (query) ->
  socket.current = query.name # Blocking 処理
  Data.find(query).exec (err, data) ->
    socket.emit 'start', query
    for d, i in data
      do (d, i, data, query) ->
        setTimeout ->
          if socket.current is query.name
            socket.emit 'data', query, data
          if i+1 is data.length
            socket.emit 'end', query
        , 4 * i # 4msでデータ順序が保たれる