root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Remove Stream#toBlockingQueue use

This commit is contained in:
Stephane Maldini 2016-02-08 00:25:52 +00:00
parent 4a24979181
commit b7b423a003
1 changed files with 9 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
spring-web-reactive/src/main/java/org/springframework/core/io/buffer/support/DataBufferPublisherInputStream.java
View File

@ -18,12 +18,12 @@ package org.springframework.core.io.buffer.support;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.util.concurrent.BlockingQueue; import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription; import org.reactivestreams.Subscription;
import reactor.rx.Stream; import reactor.core.publisher.Flux;
import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.util.Assert; import org.springframework.util.Assert;
@ -35,7 +35,7 @@ class DataBufferPublisherInputStream extends InputStream {
private final AtomicBoolean completed = new AtomicBoolean(); private final AtomicBoolean completed = new AtomicBoolean();
private final BlockingQueue<DataBuffer> queue; private final Iterator<DataBuffer> queue;
private InputStream currentStream; private InputStream currentStream;
@ -57,8 +57,7 @@ class DataBufferPublisherInputStream extends InputStream {
int requestSize) { int requestSize) {
Assert.notNull(publisher, "'publisher' must not be null"); Assert.notNull(publisher, "'publisher' must not be null");
// TODO Avoid using Reactor Stream, it should not be a mandatory dependency of Spring Reactive this.queue = Flux.from(publisher).toIterable(requestSize).iterator();
this.queue = Stream.from(publisher).toBlockingQueue(requestSize);
} }
@Override @Override
@ -126,25 +125,22 @@ class DataBufferPublisherInputStream extends InputStream {
return this.currentStream; return this.currentStream;
} }
else { else {
// take() blocks until next or complete() then return null, // if upstream Publisher has completed, then complete() and return null,
// but that's OK since this is a *blocking* InputStream if (!this.queue.hasNext()) {
DataBuffer signal = this.queue.take();
if (signal == null) {
this.completed.set(true); this.completed.set(true);
return null; 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(); this.currentStream = signal.asInputStream();
return this.currentStream; return this.currentStream;
} }
} }
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
}
catch (Throwable error) { catch (Throwable error) {
this.completed.set(true); this.completed.set(true);
throw new IOException(error); throw new IOException(error);
} }
throw new IOException();
} }