 @$push_to_list    = ( collector ) -> @_map_errors ( data ) => collector.push  data; return data
 @$add_to_set      = ( collector ) -> @_map_errors ( data ) => collector.add   data; return data
 #...........................................................................................................
-@$count           = -> throw new Error "not implemented"
+@$count           = -> throw new Error "µ23183 not implemented"
 @$take            = $take
 
 #-----------------------------------------------------------------------------------------------------------

### TAINT refactor: `PS.new_source.from_path`, `PS.new_source.from_text`..., `PS.new_sink.as_text` (???) ###
@new_text_source = ( text ) -> $values [ text, ]
@new_text_sink = ->
  throw new Error "not implemented"

#-----------------------------------------------------------------------------------------------------------
@new_value_source = ( values ) -> $values values

#-----------------------------------------------------------------------------------------------------------
@map_start = ( method ) ->
  throw new Error "expected a function, got a #{type}" unless ( type = CND.type_of method ) is 'function'
  throw new Error "method arity #{arity} not implemented" unless ( arity = method.length ) is 0
  is_first = yes
  return @_map_errors ( data ) =>
    if is_first

  R.add = ( emitter ) ->
    ### TAINT only works with PipeStreams event emitters; could overwrite `emit` method otherwise ###
    unless ( type = CND.type_of emitter.on ) is 'function'
      throw new Error "µ4823 expected an event emitter with an `on` method, got a #{type}"
    throw new Error "µ5588 got duplicate emitter" if R._emitters.has emitter
    R._emitters.set emitter, 1
    R._source_count += +1
    emitter.on '*', aggregator
