sink termination propagation

[ronag]

I've noticed a lot of the callbag libraries out there don't propagate termination backwards to sources, which can leave dangling resources... maybe we could clarify this part of the protocol?

e.g. I have a callbag source that reads from a database. However, if there is an error further down along the callbag chain, or if the sink is terminated. The termination signal needs to be propagated back in order to free resources. Which is commonly not the case with existing callbag elements.

 (start, sink) => {
    if (start !== 0) {
      return
    }

    let changes = null
    let paused = true
    let queue = []

    sink(0, (t, d) => {
      if (t === 1) {
        if (queue.length > 0) {
          sink(1, queue.shift())
          paused = true
        } else {
          paused = false
        }

        if (!changes && queue.length < 512) {
          changes = db
            .changes({
              since,
              live: true,
              retry: true
            })
            .on('change', change => {
              since = change.seq
              if (paused) {
                sink(1, change)
                paused = true
              } else {
                queue.push(change)
              }

              if (queue.length > 1024) {
                changes.cancel()
                changes = null
              }
            })
            .on('error', err => {
              sink(2, err)
            })
        }
      } else if (t === 2) {
        // NOTE: Early termination...
        if (changes) {
          changes.cancel()
          changes = null
        }
      }
    })