DATAMONGO-1855 - Add reactive GridFS support.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>2.2.0.BUILD-SNAPSHOT</version>
+	<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-cross-store/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -50,7 +50,7 @@
 		<dependency>
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-mongodb</artifactId>
-			<version>2.2.0.BUILD-SNAPSHOT</version>
+			<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 		</dependency>
 
 		<!-- reactive -->

spring-data-mongodb-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>2.2.0.BUILD-SNAPSHOT</version>
+		<version>2.2.0.DATAMONGO-1855-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/AsyncInputStreamAdapter.java

@@ -0,0 +1,257 @@
+/*
+ * Copyright 2019 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.data.mongodb.gridfs;
+
+import lombok.RequiredArgsConstructor;
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Operators;
+import reactor.util.concurrent.Queues;
+import reactor.util.context.Context;
+
+import java.nio.ByteBuffer;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.function.BiConsumer;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+
+import com.mongodb.reactivestreams.client.Success;
+import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream;
+
+/**
+ * Adapter accepting a binary stream {@link Publisher} and emitting its through {@link AsyncInputStream}.
+ * <p>
+ * This adapter subscribes to the binary {@link Publisher} as soon as the first chunk gets {@link #read(ByteBuffer)
+ * requested}. Requests are queued and binary chunks are requested from the {@link Publisher}. As soon as the
+ * {@link Publisher} emits items, chunks are provided to the read request which completes the number-of-written-bytes
+ * {@link Publisher}.
+ * <p>
+ * {@link AsyncInputStream} is supposed to work as sequential callback API that is called until reaching EOF.
+ * {@link #close()} is propagated as cancellation signal to the binary {@link Publisher}.
+ *
+ * @author Mark Paluch
+ * @author Christoph Strobl
+ * @since 2.2
+ */
+@RequiredArgsConstructor
+class AsyncInputStreamAdapter implements AsyncInputStream {
+
+	private static final AtomicLongFieldUpdater<AsyncInputStreamAdapter> DEMAND = AtomicLongFieldUpdater
+			.newUpdater(AsyncInputStreamAdapter.class, "demand");
+
+	private static final AtomicIntegerFieldUpdater<AsyncInputStreamAdapter> SUBSCRIBED = AtomicIntegerFieldUpdater
+			.newUpdater(AsyncInputStreamAdapter.class, "subscribed");
+
+	private static final int SUBSCRIPTION_NOT_SUBSCRIBED = 0;
+	private static final int SUBSCRIPTION_SUBSCRIBED = 1;
+
+	private final Publisher<? extends DataBuffer> buffers;
+	private final Context subscriberContext;
+	private final DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+
+	private volatile Subscription subscription;
+	private volatile boolean cancelled;
+	private volatile boolean complete;
+	private volatile Throwable error;
+	private final Queue<BiConsumer<DataBuffer, Integer>> readRequests = Queues.<BiConsumer<DataBuffer, Integer>> small()
+			.get();
+
+	// see DEMAND
+	volatile long demand;
+
+	// see SUBSCRIBED
+	volatile int subscribed = SUBSCRIPTION_NOT_SUBSCRIBED;
+
+	/*
+	 * (non-Javadoc)
+	 * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#read(java.nio.ByteBuffer)
+	 */
+	@Override
+	public Publisher<Integer> read(ByteBuffer dst) {
+
+		return Mono.create(sink -> {
+
+			readRequests.offer((db, bytecount) -> {
+
+				try {
+
+					if (error != null) {
+
+						sink.error(error);
+						return;
+					}
+
+					if (bytecount == -1) {
+
+						sink.success(-1);
+						return;
+					}
+
+					ByteBuffer byteBuffer = db.asByteBuffer();
+					int toWrite = byteBuffer.remaining();
+
+					dst.put(byteBuffer);
+					sink.success(toWrite);
+
+				} catch (Exception e) {
+					sink.error(e);
+				} finally {
+					DataBufferUtils.release(db);
+				}
+			});
+
+			request(1);
+		});
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#close()
+	 */
+	@Override
+	public Publisher<Success> close() {
+
+		return Mono.create(sink -> {
+
+			cancelled = true;
+
+			if (error != null) {
+				sink.error(error);
+				return;
+			}
+
+			sink.success(Success.SUCCESS);
+		});
+	}
+
+	protected void request(int n) {
+
+		if (complete) {
+
+			terminatePendingReads();
+			return;
+		}
+
+		Operators.addCap(DEMAND, this, n);
+
+		if (SUBSCRIBED.get(this) == SUBSCRIPTION_NOT_SUBSCRIBED) {
+
+			if (SUBSCRIBED.compareAndSet(this, SUBSCRIPTION_NOT_SUBSCRIBED, SUBSCRIPTION_SUBSCRIBED)) {
+				buffers.subscribe(new DataBufferCoreSubscriber());
+			}
+
+		} else {
+
+			Subscription subscription = this.subscription;
+
+			if (subscription != null) {
+				requestFromSubscription(subscription);
+			}
+		}
+	}
+
+	void requestFromSubscription(Subscription subscription) {
+
+		long demand = DEMAND.get(AsyncInputStreamAdapter.this);
+
+		if (cancelled) {
+			subscription.cancel();
+		}
+
+		if (demand > 0 && DEMAND.compareAndSet(AsyncInputStreamAdapter.this, demand, demand - 1)) {
+			subscription.request(1);
+		}
+	}
+
+	/**
+	 * Terminates pending reads with empty buffers.
+	 */
+	void terminatePendingReads() {
+
+		BiConsumer<DataBuffer, Integer> readers;
+
+		while ((readers = readRequests.poll()) != null) {
+			readers.accept(factory.wrap(new byte[0]), -1);
+		}
+	}
+
+	private class DataBufferCoreSubscriber implements CoreSubscriber<DataBuffer> {
+
+		@Override
+		public Context currentContext() {
+			return AsyncInputStreamAdapter.this.subscriberContext;
+		}
+
+		@Override
+		public void onSubscribe(Subscription s) {
+
+			AsyncInputStreamAdapter.this.subscription = s;
+
+			Operators.addCap(DEMAND, AsyncInputStreamAdapter.this, -1);
+			s.request(1);
+		}
+
+		@Override
+		public void onNext(DataBuffer dataBuffer) {
+
+			if (cancelled || complete) {
+				DataBufferUtils.release(dataBuffer);
+				Operators.onNextDropped(dataBuffer, AsyncInputStreamAdapter.this.subscriberContext);
+				return;
+			}
+
+			BiConsumer<DataBuffer, Integer> poll = AsyncInputStreamAdapter.this.readRequests.poll();
+
+			if (poll == null) {
+
+				DataBufferUtils.release(dataBuffer);
+				Operators.onNextDropped(dataBuffer, AsyncInputStreamAdapter.this.subscriberContext);
+				subscription.cancel();
+				return;
+			}
+
+			poll.accept(dataBuffer, dataBuffer.readableByteCount());
+
+			requestFromSubscription(subscription);
+		}
+
+		@Override
+		public void onError(Throwable t) {
+
+			if (AsyncInputStreamAdapter.this.cancelled || AsyncInputStreamAdapter.this.complete) {
+				Operators.onErrorDropped(t, AsyncInputStreamAdapter.this.subscriberContext);
+				return;
+			}
+
+			AsyncInputStreamAdapter.this.error = t;
+			AsyncInputStreamAdapter.this.complete = true;
+			terminatePendingReads();
+		}
+
+		@Override
+		public void onComplete() {
+
+			AsyncInputStreamAdapter.this.complete = true;
+			terminatePendingReads();
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdapters.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2019 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.data.mongodb.gridfs;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.reactivestreams.Publisher;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+
+import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream;
+
+/**
+ * Utility methods to create adapters from between {@link Publisher} of {@link DataBuffer} and {@link AsyncInputStream}.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+class BinaryStreamAdapters {
+
+	/**
+	 * Creates a {@link Flux} emitting {@link DataBuffer} by reading binary chunks from {@link AsyncInputStream}.
+	 * Publisher termination (completion, error, cancellation) closes the {@link AsyncInputStream}.
+	 * <p/>
+	 * The resulting {@link org.reactivestreams.Publisher} filters empty binary chunks and uses {@link DataBufferFactory}
+	 * settings to determine the chunk size.
+	 *
+	 * @param inputStream must not be {@literal null}.
+	 * @param dataBufferFactory must not be {@literal null}.
+	 * @return {@link Flux} emitting {@link DataBuffer}s.
+	 * @see DataBufferFactory#allocateBuffer()
+	 */
+	static Flux<DataBuffer> toPublisher(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory) {
+
+		return DataBufferPublisherAdapter.createBinaryStream(inputStream, dataBufferFactory) //
+				.filter(it -> {
+
+					if (it.readableByteCount() == 0) {
+						DataBufferUtils.release(it);
+						return false;
+					}
+					return true;
+				});
+	}
+
+	/**
+	 * Creates a {@link Mono} emitting a {@link AsyncInputStream} to consume a {@link Publisher} emitting
+	 * {@link DataBuffer} and exposing the binary stream through {@link AsyncInputStream}. {@link DataBuffer}s are
+	 * released by the adapter during consumption.
+	 * <p/>
+	 * This method returns a {@link Mono} to retain the {@link reactor.util.context.Context subscriber context}.
+	 *
+	 * @param dataBuffers must not be {@literal null}.
+	 * @return {@link Mono} emitting {@link AsyncInputStream}.
+	 * @see DataBufferUtils#release(DataBuffer)
+	 */
+	static Mono<AsyncInputStream> toAsyncInputStream(Publisher<? extends DataBuffer> dataBuffers) {
+
+		return Mono.create(sink -> {
+			sink.success(new AsyncInputStreamAdapter(dataBuffers, sink.currentContext()));
+		});
+	}
+}