From 5519428d43e797742d60e76fcd10dab181cfcfe7 Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Thu, 17 Mar 2016 17:08:19 +0100 Subject: [PATCH] Replaced DataBufferPublisherInputStream with JDK equivalent. --- .../DataBufferPublisherInputStream.java | 147 ------------------ .../io/buffer/support/DataBufferUtils.java | 31 +++- 2 files changed, 30 insertions(+), 148 deletions(-) delete mode 100644 spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java deleted file mode 100644 index 5405b9b65b..0000000000 --- a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java +++ /dev/null @@ -1,147 +0,0 @@ -/* - * Copyright 2002-2016 the original author or authors. - * - * 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 org.springframework.core.io.buffer.support; - -import java.io.IOException; -import java.io.InputStream; -import java.util.Iterator; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscription; -import reactor.core.publisher.Flux; - -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.util.Assert; - -/** - * @author Arjen Poutsma - */ -class DataBufferPublisherInputStream extends InputStream { - - private final AtomicBoolean completed = new AtomicBoolean(); - - private final Iterator queue; - - private InputStream currentStream; - - /** - * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher. - * @param publisher the publisher to use - */ - public DataBufferPublisherInputStream(Publisher publisher) { - this(publisher, 1); - } - - /** - * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher. - * @param publisher the publisher to use - * @param requestSize the {@linkplain Subscription#request(long) request size} to use - * on the publisher bound to Integer MAX - */ - public DataBufferPublisherInputStream(Publisher publisher, - int requestSize) { - Assert.notNull(publisher, "'publisher' must not be null"); - - this.queue = Flux.from(publisher).toIterable(requestSize).iterator(); - } - - @Override - public int available() throws IOException { - if (completed.get()) { - return 0; - } - InputStream is = currentStream(); - return is != null ? is.available() : 0; - } - - @Override - public int read() throws IOException { - if (completed.get()) { - return -1; - } - InputStream is = currentStream(); - while (is != null) { - int ch = is.read(); - if (ch != -1) { - return ch; - } - else { - is = currentStream(); - } - } - return -1; - } - - @Override - public int read(byte[] b, int off, int len) throws IOException { - if (completed.get()) { - return -1; - } - InputStream is = currentStream(); - if (is == null) { - return -1; - } - else if (b == null) { - throw new NullPointerException(); - } - else if (off < 0 || len < 0 || len > b.length - off) { - throw new IndexOutOfBoundsException(); - } - else if (len == 0) { - return 0; - } - do { - int n = is.read(b, off, len); - if (n > 0) { - return n; - } - else { - is = currentStream(); - } - } - while (is != null); - - return -1; - } - - private InputStream currentStream() throws IOException { - try { - if (this.currentStream != null && this.currentStream.available() > 0) { - return this.currentStream; - } - else { - // if upstream Publisher has completed, then complete() and return null, - if (!this.queue.hasNext()) { - this.completed.set(true); - return null; - } - // next() blocks until next - // but that's OK since this is a *blocking* InputStream - DataBuffer signal = this.queue.next(); - this.currentStream = signal.asInputStream(); - return this.currentStream; - } - } - catch (Throwable error) { - this.completed.set(true); - throw new IOException(error); - } - } - - -} diff --git a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java index d4bcc1ad6c..5888ce8f34 100644 --- a/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java +++ b/spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferUtils.java @@ -17,6 +17,9 @@ package org.springframework.core.io.buffer.support; import java.io.InputStream; +import java.io.SequenceInputStream; +import java.util.Enumeration; +import java.util.Iterator; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; @@ -62,7 +65,33 @@ public abstract class DataBufferUtils { * @return the input stream */ public static InputStream toInputStream(Publisher publisher) { - return new DataBufferPublisherInputStream(publisher); + Iterable streams = Flux.from(publisher). + map(DataBuffer::asInputStream). + toIterable(); + + Enumeration enumeration = + new IteratorEnumeration(streams); + + return new SequenceInputStream(enumeration); + } + + private static class IteratorEnumeration implements Enumeration { + + private final Iterator delegate; + + public IteratorEnumeration(Iterable iterable) { + this.delegate = iterable.iterator(); + } + + @Override + public boolean hasMoreElements() { + return delegate.hasNext(); + } + + @Override + public T nextElement() { + return delegate.next(); + } } }