Channels need ability to specify "disposing" policy for their elements

[elizarov]

Consider a case where you use a Channel to send some resources that need to be disposed (consumed/closed) after use. For example, you are sending network connections, open files, other channels, handles to native memory, etc. See ReceiveChannel.flatMap, for example, that specifically works on channel of channels. It all works fine as long as the consumer receives and process all the sent elements. However, if the consumer is cancelled, then it typically cancels the channel it is receiving from and this leaves us in a state where channel's buffer can have some items that were not received yet. Currently all those items are simply dropped. This affects even RendezvousChannel as the producer might have already invoked send(element) and when consumer cancels the channel this element is just dropped.

Currently now there is no way to specify how those dropped items shall be be closed/disposed.

[venkatperi]

Perhaps call close() on Closeable messages?

Although, without active state tracking one can't be absolutely sure that a message needs to be disposed. Suppose a channel is making a decision to dispose messages, but further upstream messages have been tee'd off to another channel. Disposing a message can have unintended side effects.

fun <T> ReceiveChannel<T>.tee(listener: SendChannel<T> ) =
  produce {
    consumeEach {
      listener.send(it)
      send(it)
    }
    listener.close()
  }

[qwwdfsad]

Implementation note: it can piggyback on #341 storage

[qwwdfsad]

Fixed with onUndeliveredElement since 1.5.0