@@ -0,0 +1,1631 @@
+package de.openknowledge.jaxrs.reactive.test;%0A%0Aimport java.io.IOException;%0Aimport java.nio.ByteBuffer;%0Aimport java.nio.channels.AsynchronousFileChannel;%0Aimport java.nio.channels.CompletionHandler;%0Aimport java.util.concurrent.Flow.Subscriber;%0Aimport java.util.concurrent.Flow.Subscription;%0Aimport java.util.concurrent.Phaser;%0A%0Apublic class AsynchronousFileChannelSubscriber implements Subscriber%3CByteBuffer%3E %7B%0A%0A  private AsynchronousFileChannel channel;%0A  private long filePosition;%0A  private Subscription subscription;%0A  private Phaser phaser = new Phaser(1);%0A%0A  public AsynchronousFileChannelSubscriber(AsynchronousFileChannel fileChannel) %7B%0A    channel = fileChannel;%0A  %7D%0A%0A  @Override%0A  public void onSubscribe(Subscription s) %7B%0A    subscription = s;%0A    subscription.request(1);%0A  %7D%0A%0A  @Override%0A  public void onNext(ByteBuffer buffer) %7B%0A    phaser.register();%0A    %0A    channel.write(buffer, filePosition, filePosition, new CompletionHandler%3CInteger, Long%3E() %7B%0A%0A      @Override%0A      public void completed(Integer writeCount, Long position) %7B%0A        filePosition += writeCount;%0A        phaser.arriveAndDeregister();%0A        subscription.request(1);%0A      %7D%0A%0A      @Override%0A      public void failed(Throwable error, Long position) %7B%0A        phaser.arriveAndDeregister();%0A        subscription.cancel();%0A      %7D%0A    %7D);%0A  %7D%0A%0A  @Override%0A  public void onComplete() %7B%0A    try %7B%0A      phaser.arriveAndAwaitAdvance();%0A      phaser.arriveAndDeregister();%0A      channel.close();%0A    %7D catch (IOException e) %7B%0A      // ignore%0A    %7D%0A  %7D%0A%0A  @Override%0A  public void onError(Throwable error) %7B%0A    // TODO log error%0A    onComplete();%0A  %7D%0A%7D%0A

Add AsynchronousFileChannelSubscriber