Multi from InputStream

common/reactive/src/main/java/io/helidon/common/reactive/IoMulti.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.helidon.common.reactive;
+
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+
+import io.helidon.common.Builder;
+
+/**
+ * Create reactive stream from standard IO resources.
+ */
+public interface IoMulti {
+
+    /**
+     * Create a {@link Multi} instance that publishes {@link ByteBuffer}s from the given {@link InputStream}.
+     * <p>
+     * {@link InputStream} is trusted not to block on read operations, in case it can't be assured use
+     * builder to specify executor for asynchronous waiting for blocking reads.
+     * {@code IoMulti.builder(is).executor(executorService).build()}.
+     *
+     * @param inputStream the Stream to publish
+     * @return Multi
+     * @throws NullPointerException if {@code stream} is {@code null}
+     */
+    static Multi<ByteBuffer> create(final InputStream inputStream) {
+        return IoMulti.builder(inputStream)
+                .build();
+    }
+
+    /**
+     * Creates a builder of the {@link Multi} from supplied {@link java.io.InputStream}.
+     *
+     * @param inputStream the Stream to publish
+     * @return the builder
+     */
+    static MultiFromInputStreamBuilder builder(final InputStream inputStream) {
+        Objects.requireNonNull(inputStream);
+        return new MultiFromInputStreamBuilder(inputStream);
+    }
+
+    final class MultiFromInputStreamBuilder implements Builder<Multi<ByteBuffer>> {
+
+        private int bufferSize = 1024;
+        private ExecutorService executor;
+        private final InputStream inputStream;
+
+        MultiFromInputStreamBuilder(final InputStream inputStream) {
+            this.inputStream = inputStream;
+        }
+
+        /**
+         * Set the size of {@link ByteBuffer}s being published.
+         *
+         * @param bufferSize size of the {@link ByteBuffer}
+         * @return Multi
+         */
+        public MultiFromInputStreamBuilder byteBufferSize(int bufferSize) {
+            this.bufferSize = bufferSize;
+            return this;
+        }
+
+        /**
+         * If the {@code InputStream} can block in read method, use executor for asynchronous waiting.
+         *
+         * @param executor used for asynchronous waiting for blocking reads
+         * @return this builder
+         */
+        public MultiFromInputStreamBuilder executor(final ExecutorService executor) {
+            Objects.requireNonNull(executor);
+            this.executor = executor;
+            return this;
+        }
+
+        @Override
+        public Multi<ByteBuffer> build() {
+            if (executor != null) {
+                return new MultiFromBlockingInputStream(inputStream, bufferSize, executor);
+            }
+            return new MultiFromInputStream(inputStream, bufferSize);
+        }
+    }
+}

common/reactive/src/main/java/io/helidon/common/reactive/MultiFromBlockingInputStream.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.common.reactive;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Flow;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.IntSupplier;
+
+class MultiFromBlockingInputStream extends MultiFromInputStream {
+
+    private final InputStream inputStream;
+    private IntSupplier bufferSizeSupplier;
+    private final ExecutorService executorService;
+
+    MultiFromBlockingInputStream(InputStream inputStream, int bufferSize, ExecutorService executorService) {
+        super(inputStream, bufferSize);
+        this.inputStream = inputStream;
+        this.bufferSizeSupplier = () -> bufferSize;
+        this.executorService = executorService;
+    }
+
+    @Override
+    public void subscribe(final Flow.Subscriber<? super ByteBuffer> subscriber) {
+        Objects.requireNonNull(subscriber, "subscriber is null");
+        try {
+            inputStream.available();
+        } catch (IOException e) {
+            subscriber.onSubscribe(EmptySubscription.INSTANCE);
+            subscriber.onError(e);
+            return;
+        }
+        InputStreamSubscription subscription = new InputStreamSubscription(
+                subscriber,
+                inputStream,
+                bufferSizeSupplier.getAsInt(),
+                executorService);
+        subscriber.onSubscribe(subscription);
+    }
+
+    @Override
+    public Multi<ByteBuffer> withByteBufferSize(final int bufferSize) {
+        this.bufferSizeSupplier = () -> bufferSize;
+        return this;
+    }
+
+    static final class InputStreamSubscription extends MultiFromInputStream.InputStreamSubscription {
+
+        private final ExecutorService executorService;
+        private final LinkedBlockingQueue<Runnable> submitQueue = new LinkedBlockingQueue<>();
+
+        private final AtomicBoolean draining = new AtomicBoolean(false);
+
+        InputStreamSubscription(Flow.Subscriber<? super ByteBuffer> downstream,
+                                InputStream inputStream,
+                                final int bufferSize,
+                                ExecutorService executorService) {
+            super(downstream, inputStream, bufferSize);
+            this.executorService = executorService;
+        }
+
+        protected void trySubmit(long n) {
+            submitQueue.add(() -> {
+                submit(n);
+                drainSubmitQueue();
+            });
+            drainSubmitQueue();
+        }
+
+        private void drainSubmitQueue() {
+            if (!draining.getAndSet(true)) {
+                try {
+                    Runnable job = submitQueue.poll();
+                    if (job != null) {
+                        executorService.submit(job);
+                    }
+                } finally {
+                    draining.set(false);
+                }
+            }
+        }
+    }
+}

common/reactive/src/main/java/io/helidon/common/reactive/MultiFromInputStream.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright (c)  2020 Oracle and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.helidon.common.reactive;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.IntSupplier;
+
+class MultiFromInputStream implements Multi<ByteBuffer> {
+
+    private final InputStream inputStream;
+    private IntSupplier bufferSizeSupplier;
+
+    MultiFromInputStream(InputStream inputStream, int bufferSize) {
+        this.inputStream = inputStream;
+        this.bufferSizeSupplier = () -> bufferSize;
+    }
+
+    @Override
+    public void subscribe(final Flow.Subscriber<? super ByteBuffer> subscriber) {
+        Objects.requireNonNull(subscriber, "subscriber is null");
+        try {
+            inputStream.available();
+        } catch (IOException e) {
+            subscriber.onSubscribe(EmptySubscription.INSTANCE);
+            subscriber.onError(e);
+            return;
+        }
+        InputStreamSubscription subscription = new InputStreamSubscription(
+                subscriber,
+                inputStream,
+                bufferSizeSupplier.getAsInt());
+        subscriber.onSubscribe(subscription);
+    }
+
+    public Multi<ByteBuffer> withByteBufferSize(final int bufferSize) {
+        this.bufferSizeSupplier = () -> bufferSize;
+        return this;
+    }
+
+    static class InputStreamSubscription extends AtomicLong implements Flow.Subscription {
+
+        private final Flow.Subscriber<? super ByteBuffer> downstream;
+        private final int bufferSize;
+        private InputStream inputStream;
+
+        private volatile int canceled;
+
+        static final int NORMAL_CANCEL = 1;
+        static final int BAD_REQUEST = 2;
+
+        InputStreamSubscription(Flow.Subscriber<? super ByteBuffer> downstream,
+                                InputStream inputStream,
+                                final int bufferSize) {
+            this.downstream = downstream;
+            this.inputStream = inputStream;
+            this.bufferSize = bufferSize;
+        }
+
+        protected void submit(long n) {
+            long emitted = 0L;
+            Flow.Subscriber<? super ByteBuffer> downstream = this.downstream;
+
+            for (;;) {
+                while (emitted != n) {
+                    int isCanceled = canceled;
+                    if (isCanceled != 0) {
+                        inputStream = null;
+                        if (isCanceled == BAD_REQUEST) {
+                            downstream.onError(new IllegalArgumentException(
+                                    "Rule §3.9 violated: non-positive request amount is forbidden"));
+                        }
+                        return;
+                    }
+
+                    ByteBuffer value;
+
+                    try {
+                        value = ByteBuffer.wrap(inputStream.readNBytes(bufferSize));
+                    } catch (Throwable ex) {
+                        inputStream = null;
+                        canceled = NORMAL_CANCEL;
+                        downstream.onError(ex);
+                        return;
+                    }
+
+                    if (value.limit() == 0) {
+                        inputStream = null;
+                        downstream.onComplete();
+                        return;
+                    }
+
+                    downstream.onNext(value);
+
+                    if (canceled != 0) {
+                        continue;
+                    }
+
+                    emitted++;
+                }
+
+                n = get();
+                if (n == emitted) {
+                    n = SubscriptionHelper.produced(this, emitted);
+                    if (n == 0L) {
+                        return;
+                    }
+                    emitted = 0L;
+                }
+            }
+        }
+
+        @Override
+        public void request(long n) {
+            if (n <= 0L) {
+                canceled = BAD_REQUEST;
+                n = 1; // for cleanup
+            }
+
+            if (SubscriptionHelper.addRequest(this, n) != 0L) {
+                return;
+            }
+
+            trySubmit(n);
+        }
+
+        protected void trySubmit(long n) {
+            submit(n);
+        }
+
+        @Override
+        public void cancel() {
+            canceled = NORMAL_CANCEL;
+            request(1); // for cleanup
+        }
+    }
+}