content: package hu.akarnokd.rxjava2;
  
  import java.io.*;
  import java.nio.charset.Charset;
  import java.util.concurrent.atomic.AtomicReference;
  
  import org.reactivestreams.*;
  
  import io.reactivex.FlowableSubscriber;
  import io.reactivex.internal.subscriptions.SubscriptionHelper;
  
  public final class FlowableStringInputStream {
  
      private FlowableStringInputStream() {
          throw new IllegalStateException("No instances!");
      }
  
      public static InputStream createInputStream(Publisher<String> source, Charset charset) {
          StringInputStream parent = new StringInputStream(charset);
          source.subscribe(parent);
          return parent;
      }
  
      static final class StringInputStream extends InputStream
      implements FlowableSubscriber<String> {
  
          final AtomicReference<Subscription> upstream;
  
          final Charset charset;
  
          volatile byte[] bytes;
  
          int index;
  
          volatile boolean done;
          Throwable error;
  
          StringInputStream(Charset charset) {
              this.charset = charset;
              upstream = new AtomicReference<>();
          }
  
          @Override
          public void onSubscribe(Subscription s) {
              if (SubscriptionHelper.setOnce(upstream, s)) {
                  s.request(1);
              }
          }
  
          @Override
          public void onNext(String t) {
              bytes = t.getBytes(charset);
              synchronized (this) {
                  notifyAll();
              }
          }
  
          @Override
          public void onError(Throwable t) {
              error = t;
              done = true;
              synchronized (this) {
                  notifyAll();
              }
          }
  
          @Override
          public void onComplete() {
              done = true;
              synchronized (this) {
                  notifyAll();
              }
          }
  
          @Override
          public int read() throws IOException {
              for (;;) {
                  byte[] a = awaitBufferIfNecessary();
                  if (a == null) {
                      Throwable ex = error;
                      if (ex != null) {
                          if (ex instanceof IOException) {
                              throw (IOException)ex;
                          }
                          throw new IOException(ex);
                      }
                      return -1;
                  }
                  int idx = index;
                  if (idx == a.length) {
                      index = 0;
                      bytes = null;
                      upstream.get().request(1);
                  } else {
                      int result = a[idx] & 0xFF;
                      index = idx + 1;
                      return result;
                  }
              }
          }
  
          byte[] awaitBufferIfNecessary() throws IOException {
              byte[] a = bytes;
              if (a == null) {
                  synchronized (this) {
                      for (;;) {
                          boolean d = done;
                          a = bytes;
                          if (a != null) {
                              break;
                          }
                          if (d || upstream.get() == SubscriptionHelper.CANCELLED) {
                              break;
                          }
                          try {
                              wait();
                          } catch (InterruptedException ex) {
                              if (upstream.get() != SubscriptionHelper.CANCELLED) {
                                  InterruptedIOException exc = new InterruptedIOException();
                                  exc.initCause(ex);
                                  throw exc;
                              }
                              break;
                          }
                      }
                  }
              }
              return a;
          }
  
          @Override
          public int read(byte[] b, int off, int len) throws IOException {
              if (off < 0 || len < 0 || off >= b.length || off + len > b.length) {
                  throw new IndexOutOfBoundsException("b.length=" + b.length + ", off=" + off + ", len=" + len);
              }
              for (;;) {
                  byte[] a = awaitBufferIfNecessary();
                  if (a == null) {
                      Throwable ex = error;
                      if (ex != null) {
                          if (ex instanceof IOException) {
                              throw (IOException)ex;
                          }
                          throw new IOException(ex);
                      }
                      return -1;
                  }
                  int idx = index;
                  if (idx == a.length) {
                      index = 0;
                      bytes = null;
                      upstream.get().request(1);
                  } else {
                      int r = 0;
                      while (idx < a.length && len > 0) {
                          b[off] = a[idx];
                          idx++;
                          off++;
                          r++;
                          len--;
                      }
                      index = idx;
                      return r;
                  }
              }
          }
  
          @Override
          public int available() throws IOException {
              byte[] a = bytes;
              int idx = index;
              return a != null ? Math.max(0, a.length - idx) : 0;
          }
  
          @Override
          public void close() throws IOException {
              SubscriptionHelper.cancel(upstream);
              synchronized (this) {
                  notifyAll();
              }
          }
      }
  }
  