       # that is any better than the buffer implementation in nodejs streams themselves.
       stream._read = ->
 
-      opChannel = getDocOpChannel cName, docName
       redisObserver = redisLib.createClient redis.port, redis.host, redis.options
 
       open = true

            if Math.random() < 0.4
              snapshotDb.setSnapshot cName, docName, {v:opData.v, data:doc} unless snapshotDb.closed

    # Callback called with (err, op stream)
    observe: (cName, docName, v, callback) ->
      stream = new Readable objectMode:yes

      # This function is for notifying us that the stream is empty and needs data.

        throw new Error 'Stream already closed' unless open
        stream.push null
        open = false
        redisObserver.unsubscribe channel
        redisObserver.quit()

      redisObserver.on 'message', (_channel, msg) ->
        assert open
        return unless _channel is channel
        data = JSON.parse msg
        stream.push data

      redisObserver.subscribe channel, (err) ->
        if err
          stream.destroy()
          callback err, null
        else
          callback null, stream

    # Callback called with (err, op stream). v must be in the past or present. Behaviour
    # with a future v is undefined (because I don't think thats an interesting case).
    observe: (cName, docName, v, callback) ->
      opChannel = getDocOpChannel cName, docName

      # Subscribe redis to the stream first so we don't miss out on any operations
      # while we're getting the history
      @_subscribe opChannel, (err, stream) ->
        callback err if err

        # From here on, we need to call stream.destroy() if there are errors.
        getOps cName, docName, v, (err, data) ->
          if err
            stream.destroy()
            return callback err

          # Ok, so if there's anything in the stream right now, it might overlap with the
          # historical operations. We'll pump the reader and (probably!) prefix it with the
          # getOps result.
          queue = (d while d = stream.read())

          callback null, stream

          # First send all the operations between v and when we called getOps
          for d in data
