text: var EventEmitter = require('events').EventEmitter
  var i = require('iterate')
  var duplex = require('duplex')
  var inherits = require('util').inherits
  var serializer = require('stream-serializer')
  var u = require('./util')
  var timestamp = require('monotonic-timestamp')
  
  exports =
  module.exports = Scuttlebutt
  
  exports.createID = u.createID
  exports.updateIsRecent = u.filter
  exports.filter = u.filter
  exports.timestamp = timestamp
  
  function dutyOfSubclass() {
    throw new Error('method must be implemented by subclass')
  }
  
  //  对传入的 update 的简单数据验证
  function validate (data) {
    if(!(Array.isArray(data)
      && 'string' === typeof data[2]
      && '__proto__'     !== data[2] //THIS WOULD BREAK STUFF
      && 'number' === typeof data[1]
    )) return false
  
    return true
  }
  
  inherits (Scuttlebutt, EventEmitter)
  
  function Scuttlebutt (opts) {
  
    if(!(this instanceof Scuttlebutt)) return new Scuttlebutt(opts)
    var id = 'string' === typeof opts ? opts : opts && opts.id
    // this.sources 就是本对象维护的 Vector Clock
    // 其格式为 { sourceId1: ts1, sourceId2: ts2...}
    // this.sources 的内容是根据本对象收到的 update 积累而来
    this.sources = {}
    this.setMaxListeners(Number.MAX_VALUE)
    // count how many other instances we are replicating to.
    // this._streams 维护了当前对象需要 pipe 的 Stream 的数量
    this._streams = 0
    if(opts && opts.sign && opts.verify) {
      //  缺省的 id 是一个随机数，当然你可以用诸如 uuid 之类的 id
      this.setId(opts.id || opts.createId())
  
      //  现实环境中传递消息是需要考虑“坏人”的
      // security.js 中有实现 sign 和 verify 的例子
      // 例子中，每个 source 都都拥有相同的 private/public keys，
      // 用来给发出的 update 签名，以及校验收到的 update。
      // 如果 source 一方是浏览器，上述方法就不行了，可以借助于链路层(https)安全来保证
      this._sign   = opts.sign
      this._verify = opts.verify
    } else {
      this.setId(id || u.createId())
    }
  }
  
  var sb = Scuttlebutt.prototype
  
  var emit = EventEmitter.prototype.emit
  
  // 当收到来自于其他 source 的更新数据时，子类负责将数据同步到自己的数据结构中。
  sb.applyUpdate = dutyOfSubclass
  
  // 当需要将子类所掌握的更新历史同步到其他 source 时，需要先从子类获取 history
  sb.history     = dutyOfSubclass
  
  // 在本地记录一个变更事务。trx 代表一个变更值，由子类定义其含义
  // this._update 构造一个 update 对象，然后通过事件触发该 update 在各个相连的 stream 中发送
  // 最后，通过子类的 applyUpdates 也更新子类自身的内部状态
  sb.localUpdate = function (trx) {
    this._update([trx, timestamp(), this.id])
    return this
  }
  
  // _update 方法不仅仅被本 source 的 localUpdate 调用，也受同步过来的 gossip 的影响
  // 因此 sourceId 由外部传入
  // 数据有效性通过构造的时传入的 opts._verify 常见来校验
  // update记录的格式为: [value, ts, sourceId]
  sb._update = function (update) {
    //validated when it comes into the stream
    var ts = update[1]        // 数组第二个参数是时间戳
    var source = update[2]    // Source Id
  
    //if this message is old for it's source, ignore it. it's out of
    //order. each node must emit it's changes in order!
    //emit an 'old_data' event because i'll want to track how many
    //unnecessary messages are sent.
  
    // ts 仅在一个 source 内才有意义（不同的 source 的时钟未必一致）
    // 如果发现传入的 [ts, sourceId] 比保存的 [ts, sourceId] 新，才会产生后续的更新
    // 否则触发事件后丢弃
    // 子类可以确定的是，只会收到某个 source 的最新的 update
    // 父类发出的 old_data 事件可以帮助我们跟踪有多少无效的更新被传递
    var latest = this.sources[source]
    if(latest && latest >= ts)
      return emit.call(this, 'old_data', update), false
  
    // 父类的 sources 中，为每个 source 记录了最新的时间戳
    this.sources[source] = ts
  
    var self = this
  
    // 传入的更新数据可以被校验。无论校验成功与否，都会触发对应的事件
    // 如果数据没有问题，则调用子类的 applyUpdate 完成数据的合并
    function didVerification (err, verified) {
  
      // I'm not sure how what should happen if a async verification
      // errors. if it's an key not found - that is a verification fail,
      // not a error. if it's genunie error, really you should queue and
      // try again? or replay the message later
      // -- this should be done my the security plugin though, not scuttlebutt.
  
      if(err)
        return emit.call(self, 'error', err)
  
      if(!verified)
        return emit.call(self, 'unverified_data', update)
  
      // check if this message is older than
      // the value we already have.
      // do nothing if so
      // emit an 'old_data' event because i'll want to track how many
      // unnecessary messages are sent.
  
      // 把 update 送给子类
      // 父类并不关心子类是如何根据 update 来合并其内部状态的
      // 每一个相连的 Stream 都会监听 _update 事件，一旦收到该事件，则会触发发送同步消息到对端
      if(self.applyUpdate(update))
        emit.call(self, '_update', update) //write to stream.
    }
  
    // 仅对外部传入的 update 进行校验
    // 无论外部传入的 update，还是自己的 update，都通过 didVerification
    // 调用子类的 appleyUpdates 进行内部状态更新
    // 更新后发出 _update 事件，从而触发各个 stream 的发送动作
    if(source !== this.id) {
      if(this._verify)
        // 如果子类配置了自己的 verifier，那么执行子类的
        // 子类执行完后可调用（最好如此）父类的传入的 didVerification
        this._verify(update, didVerification)
      else
        // 如果没有配置校验方法，则直接 applyUpdates。
        didVerification(null, true)
    } else {
      if(this._sign) {
        // could make this async easily enough.
        // 如果配置了签名插件，update 数组的第四个值是签名校验结果
        update[3] = this._sign(update)
      }
      didVerification(null, true)
    }
  
    return true
  }
  
  // 一个 Scuttlebutt 为每一个同步的 source 建立一个 stream
  // 以及标识该 stream 是只读，只写还是可读写的
  sb.createStream = function (opts) {
    var self = this
    // the sources for the remote end.
    // 每一个 stream 只关心对端的 sources
    var sources = {}, other
    var syncSent = false, syncRecv = false
  
    // Scuttlebutt 内部的 Stream 计数
    this._streams ++
  
    opts = opts || {}
  
    // 下面这几行中，outer 是真正返回给调用者的 stream
    // outer 封装了一个 全双工的 stream 的实现 d
    // outer 的作用是让这个全双工的 stream 采用换行符分割的 JSON 编码方式
    var d = duplex()
    d.name = opts.name
    var outer = serializer(opts && opts.wrapper)(d)
    outer.inner = d
  
    // 当前 Scuttlebutt 是否可读写
    // 双工和可读写是两个概念
    // 双工意味着 Scuttlebutt 必须尽到连接的义务，例如：生成 History 和接收或转发 didget
    // 只读则意味着其内部状态是否会随着收到的 update 而更新
    d.writable = opts.writable !== false
    d.readable = opts.readable !== false
  
    syncRecv   = !d.writable
    syncSent   = !d.readable
  
    var tail = opts.tail !== false //default to tail=true
  
    function start (data) {
      //when the digest is recieved from the other end,
      //send the history.
      //merge with the current list of sources.
      if (!data || !data.clock) {
          d.emit('error');
          return d._end()
      }
  
      sources = data.clock
  
      // d._data 方法用来发送数据
      // 下面这句是根据收到的 digest ，生成特定的 history 发送出去
      // 由于每一个 Node 手里的 sources 的状态是不同的
      i.each(self.history(sources), function (data) {d._data(data)})
  
      //the _update listener must be set after the history is queued.
      //otherwise there is a race between the first client message
      //and the next update (which may come in on another stream)
      //this problem will probably not be encountered until you have
      //thousands of scuttlebutts.
  
      // Stream 在这句之后，开始监听 Scuttlebutt 的变化
      // 只有在子类的 applyUpdates 执行完成后，该事件才会被发送
      self.on('_update', onUpdate)
  
      // 在 history 发送完成后，向对方发送 SYNC 消息，
      // 这样对端就知道哪些是 history，哪些是后续的 update
      d._data('SYNC')
      syncSent = true
      //when we have sent all history
      outer.emit('header', data)
      outer.emit('syncSent')
      //when we have recieved all history
      //emit 'synced' when this stream has synced.
      if(syncRecv) outer.emit('sync'), outer.emit('synced')
      // 结束当前的发送工作，进入 drain 状态。
      if(!tail) d._end()
    }
  
    d
      .on('_data', function (data) {
        //if it's an array, it's an update.
        if(Array.isArray(data)) {
          //check whether we are accepting writes.
          // 如果 stream 是只读的，则忽略传递进来的 update
          if(!d.writable)
            return
          if(validate(data))
            return self._update(data)
        }
        //if it's an object, it's a scuttlebut digest.
        // digest 就是 vector clock，也就是 this.sources
        else if('object' === typeof data && data)
          start(data)
        else if('string' === typeof data && data == 'SYNC') {
          syncRecv = true
          outer.emit('syncRecieved')
          if(syncSent) outer.emit('sync'), outer.emit('synced')
        }
      }).on('_end', function () {
        d._end()
      })
      .on('close', function () {
        self.removeListener('_update', onUpdate)
        self.removeListener('dispose', dispose)
        //emit the number of streams that are remaining...
        //this will be used for memory management...
        self._streams --
        emit.call(self, 'unstream', self._streams)
      })
  
    if(opts && opts.tail === false) {
      outer.on('sync', function () {
        process.nextTick(function () {
          d._end()
        })
      })
    }
    function onUpdate (update) { //value, source, ts
      // validate 完成出站校验，无论这个 update 是我自己产生的，
      // 还是其他 source 发来的，出站前都要校验
      // u.filter 则是检查，对端的 source 是否需要接收该更新
      // 这是通过确定 update 中的 [ts, sourceId] 是不是比 sources[sourceId].ts 更新来绝定的
      if(!validate(update) || !u.filter(update, sources))
        return
  
      d._data(update)
  
      //really, this should happen before emitting.
      var ts = update[1]
      var source = update[2]
      sources[source] = ts
    }
  
    function dispose () {
      d.end()
    }
  
    // Stream 建立之后，需要发送自己的摘要数据(vector clock)给对端
    // meta 的是用来在 stream 上附带信息
    var outgoing = { id : self.id, clock : self.sources }
  
    if (opts && opts.meta) outgoing.meta = opts.meta
  
    if(d.readable) {
      d._data(outgoing)
      if(!d.writable && !opts.clock)
        start({clock:{}})
  
    } else if (opts.sendClock) {
      //send my current clock.
      //so the other side knows what to send
      d._data(outgoing)
    }
  
    self.once('dispose', dispose)
  
    return outer
  }
  
  sb.createWriteStream = function (opts) {
    opts = opts || {}
    opts.writable = true; opts.readable = false
    return this.createStream(opts)
  }
  
  sb.createReadStream = function (opts) {
    opts = opts || {}
    opts.writable = false; opts.readable = true
    return this.createStream(opts)
  }
  
  sb.dispose = function () {
    emit.call(this, 'dispose')
  }
  
  sb.setId = function (id) {
    if('__proto__' === id) throw new Error('__proto__ is invalid id')
    if(id == null) throw new Error('null is not invalid id')
    this.id = id
    return this
  }
  
  function streamDone(stream, listener) {
  
    function remove () {
      stream.removeListener('end',   onDone)
      stream.removeListener('error', onDone)
      stream.removeListener('close',   onDone)
    }
    function onDone (arg) {
      remove()
      listener.call(this, arg)
    }
  
    //this makes emitter.removeListener(event, listener) still work
    onDone.listener = listener
  
    stream.on('end',   onDone)
    stream.on('error', onDone)
    stream.on('close', onDone)
  }
  
  //create another instance of this scuttlebutt,
  //that is in sync and attached to this instance.
  sb.clone = function () {
    var A = this
    var B = new (A.constructor)
    B.setId(A.id) //same id. think this will work...
  
    // 直接修改 A 的被 clone 的数量
    A._clones = (A._clones || 0) + 1
  
    // 既然不通过网络，就无需 JSON 序列化了
    var a = A.createStream({wrapper: 'raw'})
    var b = B.createStream({wrapper: 'raw'})
  
    //all updates must be sync, so make sure pause never happens.
    a.pause = b.pause = function noop(){}
  
    streamDone(b, function () {
      A._clones--
      emit.call(A, 'unclone', A._clones)
    })
  
    a.pipe(b).pipe(a)
    //resume both streams, so that the new instance is brought up to date immediately.
    a.resume()
    b.resume()
  
    return B
  }
  
  