Question on how to Cleanup on Errors in .collect().in()

[peter-mueller]

Hi, first off I have to state I really like the effort you put into making Mutiny so readable ❤️ !

My question is, if the following text describes excepted usage (or maybe I miss something?):

If i want to use my CollectorThatNeedsToBeCleanedUp (for Example delete a internal TempFile and close a session) the .collect().in() does not give me a hook to run the cleanup code on a failure and i need to fall back to https://smallrye.io/smallrye-mutiny/guides/plug.

I'm asking because a "Subscriber" Interface that automatically requests(1) after onNext would be the sweet spot for me (at least i think so) but I didn't find something like this.

---

Example: Collector that needs to be closed

Multi.createFrom().items(...)        
      .collect().in(
          CollectorThatNeedsToBeCleanedUp::new,
          CollectorThatNeedsToBeCleanedUp::add
      )
      // is there a way to do this:
      .?doWithCollectorOnFailure?(collector -> collector.close())

[jponge]

There is no such operator but you might be able to do something with the Uni returned by in(supplier, accumulator).

Or perhaps you can rewrite in a transformToUni of some kind?

[peter-mueller]

Thanks for the response!

I've tried transformToUni and arrived at two solutions:

// "Context Pattern" that now has emerged for me to access the resource everywhere in the pipeline
// (i don't feel to great about this, and would rather create the collector/context somewhere in the pipeline and 
// reuse it later but i understand that would require a `MultiWithContext<T, C (My-Context-Type)>` or something 
// with i think is out of scope of this lib. But also some Examples on the Mutiny website access final resources 
// from outside of the pipeline
final ABlockingCollector collector = ...

myMulti
  .emitOn(Infrastructure.getDefaultWorkerPool())
  .onItem().transformToUni(row -> {
      collector.add(row);
      return Uni.createFrom().item(row);
  })
  .concatenate()
  .collect().last()
  .onFailure/eventually()...(() -> collector.close())

// OR
  .emitOn(Infrastructure.getDefaultWorkerPool())
  .collect().in(
        () -> collector,
        ExcelSpreadSheetCollector::add
  )
  .onFailure/eventually()...(() -> collector.close())

[jponge]

(and thanks for the nice words!)