Flow.fromPublisher: allow releasing elements which are buffered but not emitted #528

Flow.fromPublisher buffers elements in a channel. Some of them never reach the consumer:

• when the publisher fails, channel.error drops the buffered elements
• when the consumer stops early (an exception, an interruption), the buffered elements are never received
• elements can still arrive after the subscription is cancelled

If the elements hold resources, e.g. Netty's reference-counted ByteBufs, nobody releases them, so they leak.

Proposal: an optional callback, e.g. fromPublisher(p, onDiscard: T => Unit), called for each element received from the publisher but not emitted downstream.

Possible implementation: on publisher error, store the error and call channel.done() (so buffered elements are kept). When the flow ends early, cancel the subscription, close the channel, and drain it, calling the callback. In onNext, call the callback if the channel is already closed.

Context: tapir now tracks the unconsumed elements itself, using a lock: softwaremill/tapir#5573