code: function () {
        //retrive next messages.
        //TODO: respond to back pressure from streams to each peer.
        //if a given stream is paused, don't retrive more msgs
        //for that peer/stream.
        for(var id in this.state.peers) {
          var state = this.state.peers[id]
          while(state.retrive.length) {
            var id = state.retrive.shift()
            if(state.replicating[id])
              opts.getAt({id: id, sequence:state.replicating[id].sent+1}, this._retrive)
          }
        }
        if(this.state.receive.length) {
          var ev = this.state.receive.shift()
          opts.append(ev.value, function (err, data) {
            if(err) {
              if(this.logging) console.error('EBT:err', err)
              self.block(ev.value.author, ev.id, true)
            }
            else self.onAppend(ev.value)
          })
        }
        for(var k in this.streams)
          this.streams[k].resume()
      }
docs: _append: function (err, data, peer) { if(data) { self.onAppend(data.value ? data.value : data) } else //this definitely can happen. //TODO: broadcast fork proofs console.log('error appending:', err) },