jw
map
suspend override fun subscribe() = produce(Unconfined) { for (value in upstream.subscribe()) { send(mapper(value)) } }