Reactor-core: Provide a way to release dropped message.

[hepin1989]

If a message, let's say a Netty's Pooled ByteBuf is dropped, Which require us to call io.netty.util.ReferenceCountUtil#release(msg) to release it .

Otherwise, this will be a leak.

Should I do the release by myself or ,add an overloaded method eg

public final Flux<T> filter(Predicate<? super T> p,Callable<T> cleaner)

to it?

and let the current filter implement by filter(p,noop)?

So does other methods like distinct, dropWhile etcs

[OlegDokuka]

Great idea! I suggest to have a base operator that allows processing filtered (let's name them as dropped) items in another stream, so the base API will be like next

public final Flux<T> filter(Predicate<? super T> p,Consumer<Flux<T>> dropedStream)

so it may be used as next:

  .filter(somePredicate(), dropedFlux -> dropedFlux.map(doSomethingWithEachDropped))

This gives better flexibility in items processing from my of view.

@smaldini, @simonbasle WDYT?

[akarnokd]

No need for such complications, have the predicate release on false:

buffers.filter(buf -> {
   if (check(buf)) {
      return true;
   }
   io.netty.util.ReferenceCountUtil#release(buf)
   return false;
});

You could even do a wrapper predicate:

public static <T extends ByteBuf> Predicate<T> onFalseDrop(Predicate<? super T> predicate) {
    return t -> {
        if (predicate.test(t)) {
           return true;
        }
        io.netty.util.ReferenceCountUtil#release(t)
        return false;
    };
}

[hepin1989]

~then, how about distinct?~
seems I could do that too.

But ,I have to release it every time by hand not with a reusable function:).

[mostroverkhov]

Other option is to convert to nio Bytebuffer with Bytebuf.nioBuffers() before any transformations.

[smaldini]

for this specific case the solution from @akarnokd could just do fine. In general there is a story/theme for 2018 and 3.2 to provide for drop hooks, there are situations such as premature cancel on an accumulating/buffering operator that might need this.

[smaldini]

Another angle to this to explore in the next major api version is whether we would return a FilteredFlux instead of Flux that would provide both the nominal and the else() path similar to the specific ConnectableFlux.

[hepin1989]

I do know how could I support this with the current API but I still want it to be more functional and more concise with a Cleaner like hook:).

So does when the subscription is canceled.

[smaldini]

We are pondering 2 design solutions there:

• provide a per-flux hook to react on dropped item from upstream (so works too with cancellation from user and data is aggregated in buffer() for instance) :
  ```
  flux.filter(xxx)
  .buffer()
  .onItemDrop(item -> {
  if (item instanceof ByteBuf) {
  ((ByteBuf)item).release();
  }})
  .subscribe(someSub);

- Alternatively provide a generic support if the item implement Disposable to call dispose everytime we drop an item. That would force wrapping over each item tho but it doesn't seem that bad a solution since instead of having a switch-case with lots of instanceof you now use the right contract per item type. Also not every item needs to be disposed : 
 ```
flux.map(i -> Disposables.wrap(i, ByteBuf::release))
.filter(disp -> filter(disp.getItem()))
.buffer()
.map(AutoDisposable::getItem)
.subscribe();

[simonbasle]

• buffering operators
  
  
  
  • on cancel
  
  
  • on error termination
  
  
• filtering and conditional operators (filtering includes eg. windows with gaps)
  
  
  
  • on tryNext
  
  
  • on filtering
  
  
• fuseable operators
  
  
  
  • on cancel
  
  
  • on error termination
  
  
  • on clear