org.argeo.cms/src/org/argeo/cms/util/AsyncPipedOutputStream.java

   1 package org.argeo.cms.util;
   2
   3 import java.io.IOException;
   4 import java.io.InputStream;
   5 import java.io.OutputStream;
   6 import java.io.PipedInputStream;
   7 import java.io.PipedOutputStream;
   8 import java.io.UncheckedIOException;
   9 import java.util.Objects;
  10 import java.util.concurrent.CompletableFuture;
  11 import java.util.concurrent.TimeUnit;
  12 import java.util.function.Consumer;
  13
  14 /**
  15  * An output stream whose {@link #close()} method will wait for read actions to
  16  * be completed. It is meant to be used transparently as an
  17  * {@link OutputStream}, fulfilling the expectation that everything has been
  18  * done when the {@link #close()} method has returned.
  19  */
  20 public class AsyncPipedOutputStream extends PipedOutputStream {
  21 //      private final static Logger logger = System.getLogger(AsyncPipedOutputStream.class.getName());
  22
  23         private CompletableFuture<Void> readingDone;
  24
  25         private long timeout = 60 * 1000;
  26
  27         /**
  28          * Provides the actions which will read (and close) the related piped input
  29          * stream. Reading starts immediately asynchronously, but the provided
  30          * {@link InputStream} will block until data starts to be written to this output
  31          * stream.
  32          */
  33         public void asyncRead(Consumer<InputStream> readActions) {
  34                 try {
  35                         PipedInputStream in = new PipedInputStream(this);
  36                         readingDone = CompletableFuture.runAsync(() -> {
  37                                 readActions.accept(in);
  38                         });
  39                 } catch (IOException e) {
  40                         throw new UncheckedIOException("Cannot create piped input stream", e);
  41                 }
  42         }
  43
  44         /**
  45          * Closes this output stream immediately but then wait for the reading of the
  46          * related input stream to be completed.
  47          */
  48         @Override
  49         public void close() throws IOException {
  50                 Objects.requireNonNull(readingDone, "Async read must have started");
  51                 super.flush();
  52                 super.close();
  53                 readingDone.orTimeout(timeout, TimeUnit.MILLISECONDS).join();
  54 //              logger.log(Logger.Level.DEBUG, "OUT waiting " + timeout);
  55 //              try {
  56 //                      readingDone.get(timeout, TimeUnit.MILLISECONDS);
  57 //              } catch (InterruptedException | ExecutionException | TimeoutException e) {
  58 //                      logger.log(Logger.Level.ERROR, "Reading was not completed", e);
  59 //              }
  60 //              logger.log(Logger.Level.DEBUG, "OUT closed");
  61         }
  62
  63         /**
  64          * Sets the timeout in milliseconds when waiting for reading to be completed
  65          * before returning in the {@link #close()} method.
  66          */
  67         public void setTimeout(long timeout) {
  68                 this.timeout = timeout;
  69         }
  70
  71 }