/*
 * 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.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.rx.Stream;

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 BlockingQueue<DataBuffer> queue;

	private InputStream currentStream;

	/**
	 * Creates a new {@code ByteArrayPublisherInputStream} based on the given publisher.
	 * @param publisher the publisher to use
	 */
	public DataBufferPublisherInputStream(Publisher<DataBuffer> 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<DataBuffer> publisher,
			int requestSize) {
		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 = Stream.from(publisher).toBlockingQueue(requestSize);
	}

	@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 {
				// take() blocks until next or complete() then return null,
				// but that's OK since this is a *blocking* InputStream
				DataBuffer signal = this.queue.take();
				if (signal == null) {
					this.completed.set(true);
					return null;
				}
				this.currentStream = signal.asInputStream();
				return this.currentStream;
			}
		}
		catch (InterruptedException ex) {
			Thread.currentThread().interrupt();
		}
		catch (Throwable error) {
			this.completed.set(true);
			throw new IOException(error);
		}
		throw new IOException();
	}


}

@@ -745,32 +745,16 @@
 til.
-concurrent.BlockingQueue
+Iterator
 ;%0Aim
@@ -900,17 +900,27 @@
 tor.
-rx.Stream
+core.publisher.Flux
 ;%0A%0Ai
@@ -1184,29 +1184,24 @@
 e final 
-BlockingQueue
+Iterator
 %3CDataBuf
@@ -1915,105 +1915,8 @@
 );%0A%0A
-%09%09// TODO Avoid using Reactor Stream, it should not be a mandatory dependency of Spring Reactive%0A
 %09%09th
@@ -1926,22 +1926,20 @@
 queue = 
-Stream
+Flux
 .from(pu
@@ -1953,20 +1953,15 @@
 ).to
-BlockingQueu
+Iterabl
 e(re
@@ -1970,16 +1970,27 @@
 estSize)
+.iterator()
 ;%0A%09%7D%0A%0A%09@
@@ -3204,64 +3204,197 @@
 %09// 
-take() blocks until next or complete() then return null,
+if upstream Publisher has completed, then complete() and return null,%0A%09%09%09%09if (!this.queue.hasNext()) %7B%0A%09%09%09%09%09this.completed.set(true);%0A%09%09%09%09%09return null;%0A%09%09%09%09%7D%0A%09%09%09%09// next() blocks until next
 %0A%09%09%09
@@ -3489,96 +3489,15 @@
 eue.
-take();%0A%09%09%09%09if (signal == null) %7B%0A%09%09%09%09%09this.completed.set(true);%0A%09%09%09%09%09return null;%0A%09%09%09%09%7D
+next();
 %0A%09%09%09
@@ -3586,87 +3586,8 @@
 %09%09%7D%0A
-%09%09catch (InterruptedException ex) %7B%0A%09%09%09Thread.currentThread().interrupt();%0A%09%09%7D%0A
 %09%09ca
@@ -3680,35 +3680,8 @@
 %09%09%7D%0A
-%09%09throw new IOException();%0A
 %09%7D%0A%0A

Remove Stream#toBlockingQueue use