subscribe (observer) {
    if (this._head && observer.onKeys) {
      observer.onKeys(this._head)
    }
    if (this._queuedRecords.length > 0 && observer.onNext) {
      for (let i = 0; i < this._queuedRecords.length; i++) {
        observer.onNext(this._queuedRecords[i])
        if (this._queuedRecords.length - i - 1 <= this._lowRecordWatermark) {
          this._autoPull = true
          if (this._state === _states.READY) {
            this._handleStreaming()
          }
        }
      }
    }
    if (this._tail && observer.onCompleted) {
      observer.onCompleted(this._tail)
    }
    if (this._error) {
      observer.onError(this._error)
    }
    this._observers.push(observer)

    if (this._state === _states.READY) {
      this._handleStreaming()
    }
  }
Subscribe to events with provided observer.
@param {Object} observer - Observer object
@param {function(keys: String[])} observer.onKeys - Handle stream header, field keys.
@param {function(record: Object)} observer.onNext - Handle records, one by one.
@param {function(metadata: Object)} observer.onCompleted - Handle stream tail, the metadata.
@param {function(error: Object)} observer.onError - Handle errors, should always be provided.