Problem with lambda properties and threading

[cware-architect]

First, I was having issues with the OutputToReadStream.wrap( input ) method, I was trying to perform a proper close on the stream. But you can't close the output stream until the operation is complete. Therefore, the close() must be called on the future from pipeTo(...) inside onComplete( x-> {} ).

First, is there a cleaner way to close the output stream? You can't use the try-resource block because it closes the output too soon.

final OutputToReadStream os = new OutputToReadStream( vertx );
final Future<Void> pipeTo = os.wrap( image ).onComplete( x -> {
              try {
                os.close();
                }
              catch( Exception e ) {
                e.printStackTrace();
                }
          } );

The issue with this code is that onComplete is executed back on a vertx loop thread! And because close() makes an await call, this is bad. That means extra care must be taken to call close() on a different thread, like this...

final OutputToReadStream os = new OutputToReadStream( vertx );
final Future<Void> pipeTo = os.wrap( image ).pipeTo( ctx.response() )
      .onComplete( x -> {
            // must FJP close because onComplete is vertx loop thread!
            ForkJoinPool.commonPool().submit( () -> {
              try {
                os.close();
                }
              catch( Exception e ) {
                e.printStackTrace();
                }
              } );
          } );

*** But there's an underlying bug in the push( null ) called by close(). I think it was exasperated by the use another thread. It's throwing a NPE on this line...

        if( data == null )
           endHandler.handle( null );   <--- NPE
        else
          dataHandler.handle( data );

endHandler is null. The endHandler property is only set by one method endHandler(...) and it appears there is a call to it from the .pipeTo(...) method. The PipeImpl constructor is setting it to null. This contradicts your expected empty lambda default of x->{}

I put a null check before the call to .handle( null ), but there may be other places that need this too. Or, you can reject nulls, but that will break the PipeImpl.

I chose to replace any null settings with the empty lambda x->{} so as to keep with your initialization assumptions. This code works for me...

  @Override
  public OutputToReadStream endHandler( Handler<Void> endHandler ) {
    this.endHandler = endHandler == null ? x -> {  } : endHandler;
    return this;
    }

[guss77]

is there a cleaner way to close the output stream? You can't use the try-resource block because it closes the output too soon.

Looking at it, I think the implementation of wrap() is broken because it executes the Java blocking IO in the Vert.x context - which is incorrect (as I mentioned in issue #1), and is bad in two ways:

1. wrap() is just calling transfer() but it returns before it is complete, breaking the contract of Java blocking IO.
2. I blocks the Vert.x context thread for possibly a long time, possibly preventing it from processing the pipe output.
   wrap() also promises to close() the input stream and doesn't - and it should also close the output stream. I'll have to reimplement that.

In the mean time, I think a good implementation would be:

try (final OutputToReadStream os = new OutputToReadStream( vertx ); ) {
  final Future<Void> pipeTo = os.pipeTo(ctx.response);
  image.transferTo(os);
}

You have to remember that the Java IO is blocking and handle that by spawning a (non Vert.x context) thread for all the IO (including the try-resource) The fork-join pool is probably a good idea.

But there's an underlying bug in the push( null ) called by close().
[...]
endHandler is null [...] The PipeImpl constructor is setting it to null [...] expected empty lambda default

Well, the PipeImpl c'tor is setting the handlers to a function reference - which by definition can never be null. The problem is that its result.onComplete() is setting the event handlers to null, which is supposed to fire after close(), but with multiple threads and try-resource and stuff, close() is probably being called twice (that's the only scenario that makes sense, as endHandler triggers the resetting of the handlers, so to get an NPE it must be called again - by another internal null data frame).

I've implemented your suggestion to protect the runtime assumptions (I hate null-checking during runtime - this is exactly the kind of performance waste that polymorphism is supposed to protect us from, but apparently that's Vert.x modus operandi) and also added double-close protection.

This is still in develop but I'll work a bit on the wrap() implementation and release a new version.

[guss77]

After working on wrap() for a bit I decided it either doesn't make any sense (i.e. it requires more work for the developer than it saves ) or it is trivial. So I documented the trivial use case (as I mentioned above) and implemented a more opinionated helper that is more geared towards the use case that is at the core of this issue.

I'm not 100% sure about the implementation as once you move beyond the trivial use case of doing the initialization and writing in the same blocking context (as I showed in my comment above) you'd want to close the streams in an asynchronous context (which is what the new pipeFromInput() methods do for you) and then the Java compiler will issue "possible leak" warnings. I don't see anyway around it as even if I let the developer close the stream themselves (as the issue description shows), the compiler will still not see it and issue a warning - so I haven't made a new release yet.

@cware-architect - I'd love to hear your feedback about this issue before I do a new release.