DATAMONGO-1855 - Polishing

Introduce base class to share code between imperative and reactive GridFs.

Original Pull Request: #637

Author: christophstrobl

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

@@ -49,6 +49,7 @@
  * {@link #close()} is propagated as cancellation signal to the binary {@link Publisher}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.2
  */
 @RequiredArgsConstructor
@@ -75,10 +76,10 @@ class AsyncInputStreamAdapter implements AsyncInputStream {
 			.get();
 
 	// see DEMAND
-	private volatile long demand;
+	volatile long demand;
 
 	// see SUBSCRIBED
-	private volatile int subscribed = SUBSCRIPTION_NOT_SUBSCRIBED;
+	volatile int subscribed = SUBSCRIPTION_NOT_SUBSCRIBED;
 
 	/*
 	 * (non-Javadoc)
@@ -94,20 +95,23 @@ public Publisher<Integer> read(ByteBuffer dst) {
 				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);
 
+					dst.put(byteBuffer);
 					sink.success(toWrite);
+
 				} catch (Exception e) {
 					sink.error(e);
 				} finally {
@@ -127,6 +131,7 @@ public Publisher<Integer> read(ByteBuffer dst) {
 	public Publisher<Success> close() {
 
 		return Mono.create(sink -> {
+
 			cancelled = true;
 
 			if (error != null) {
@@ -141,6 +146,7 @@ public Publisher<Success> close() {
 	protected void request(int n) {
 
 		if (complete) {
+
 			terminatePendingReads();
 			return;
 		}
@@ -150,67 +156,9 @@ protected void request(int n) {
 		if (SUBSCRIBED.get(this) == SUBSCRIPTION_NOT_SUBSCRIBED) {
 
 			if (SUBSCRIBED.compareAndSet(this, SUBSCRIPTION_NOT_SUBSCRIBED, SUBSCRIPTION_SUBSCRIBED)) {
-
-				buffers.subscribe(new CoreSubscriber<DataBuffer>() {
-
-					@Override
-					public Context currentContext() {
-						return subscriberContext;
-					}
-
-					@Override
-					public void onSubscribe(Subscription s) {
-						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, subscriberContext);
-							return;
-						}
-
-						BiConsumer<DataBuffer, Integer> poll = readRequests.poll();
-
-						if (poll == null) {
-
-							DataBufferUtils.release(dataBuffer);
-							Operators.onNextDropped(dataBuffer, subscriberContext);
-							subscription.cancel();
-							return;
-						}
-
-						poll.accept(dataBuffer, dataBuffer.readableByteCount());
-
-						requestFromSubscription(subscription);
-					}
-
-					@Override
-					public void onError(Throwable t) {
-
-						if (cancelled || complete) {
-							Operators.onErrorDropped(t, subscriberContext);
-							return;
-						}
-
-						error = t;
-						complete = true;
-						terminatePendingReads();
-					}
-
-					@Override
-					public void onComplete() {
-
-						complete = true;
-						terminatePendingReads();
-					}
-				});
+				buffers.subscribe(new DataBufferCoreSubscriber());
 			}
+
 		} else {
 
 			Subscription subscription = this.subscription;
@@ -245,4 +193,65 @@ void terminatePendingReads() {
 			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/DataBufferPublisherAdapter.java

@@ -39,6 +39,7 @@
  * Utility to adapt a {@link AsyncInputStream} to a {@link Publisher} emitting {@link DataBuffer}.
  *
  * @author Mark Paluch
+ * @author Christoph Strobl
  * @since 2.2
  */
 class DataBufferPublisherAdapter {
@@ -51,7 +52,7 @@ class DataBufferPublisherAdapter {
 	 * @param dataBufferFactory must not be {@literal null}.
 	 * @return the resulting {@link Publisher}.
 	 */
-	public static Flux<DataBuffer> createBinaryStream(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory) {
+	static Flux<DataBuffer> createBinaryStream(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory) {
 
 		State state = new State(inputStream, dataBufferFactory);
 
@@ -73,17 +74,17 @@ public static Flux<DataBuffer> createBinaryStream(AsyncInputStream inputStream,
 	@RequiredArgsConstructor
 	static class State {
 
-		static final AtomicLongFieldUpdater<State> DEMAND = AtomicLongFieldUpdater.newUpdater(State.class, "demand");
+		private static final AtomicLongFieldUpdater<State> DEMAND = AtomicLongFieldUpdater.newUpdater(State.class, "demand");
 
-		static final AtomicIntegerFieldUpdater<State> STATE = AtomicIntegerFieldUpdater.newUpdater(State.class, "state");
+		private static final AtomicIntegerFieldUpdater<State> STATE = AtomicIntegerFieldUpdater.newUpdater(State.class, "state");
 
-		static final AtomicIntegerFieldUpdater<State> READ = AtomicIntegerFieldUpdater.newUpdater(State.class, "read");
+		private static final AtomicIntegerFieldUpdater<State> READ = AtomicIntegerFieldUpdater.newUpdater(State.class, "read");
 
-		static final int STATE_OPEN = 0;
-		static final int STATE_CLOSED = 1;
+		private static final int STATE_OPEN = 0;
+		private static final int STATE_CLOSED = 1;
 
-		static final int READ_NONE = 0;
-		static final int READ_IN_PROGRESS = 1;
+		private static final int READ_NONE = 0;
+		private static final int READ_IN_PROGRESS = 1;
 
 		final AsyncInputStream inputStream;
 		final DataBufferFactory dataBufferFactory;
@@ -140,66 +141,79 @@ void emitNext(FluxSink<DataBuffer> sink) {
 			DataBuffer dataBuffer = dataBufferFactory.allocateBuffer();
 			ByteBuffer intermediate = ByteBuffer.allocate(dataBuffer.capacity());
 
-			Mono.from(inputStream.read(intermediate)).subscribe(new CoreSubscriber<Integer>() {
+			Mono.from(inputStream.read(intermediate)).subscribe(new BufferCoreSubscriber(sink, dataBuffer, intermediate));
+		}
 
-				@Override
-				public Context currentContext() {
-					return sink.currentContext();
-				}
+		private class BufferCoreSubscriber implements CoreSubscriber<Integer> {
 
-				@Override
-				public void onSubscribe(Subscription s) {
-					s.request(1);
-				}
+			private final FluxSink<DataBuffer> sink;
+			private final DataBuffer dataBuffer;
+			private final ByteBuffer intermediate;
 
-				@Override
-				public void onNext(Integer bytes) {
+			BufferCoreSubscriber(FluxSink<DataBuffer> sink, DataBuffer dataBuffer, ByteBuffer intermediate) {
 
-					if (isClosed()) {
+				this.sink = sink;
+				this.dataBuffer = dataBuffer;
+				this.intermediate = intermediate;
+			}
 
-						onReadDone();
-						DataBufferUtils.release(dataBuffer);
-						Operators.onNextDropped(dataBuffer, sink.currentContext());
-						return;
-					}
+			@Override
+			public Context currentContext() {
+				return sink.currentContext();
+			}
 
-					intermediate.flip();
-					dataBuffer.write(intermediate);
+			@Override
+			public void onSubscribe(Subscription s) {
+				s.request(1);
+			}
 
-					sink.next(dataBuffer);
+			@Override
+			public void onNext(Integer bytes) {
 
-					try {
-						if (bytes == -1) {
-							sink.complete();
-						}
-					} finally {
-						onReadDone();
-					}
+				if (isClosed()) {
+
+					onReadDone();
+					DataBufferUtils.release(dataBuffer);
+					Operators.onNextDropped(dataBuffer, sink.currentContext());
+					return;
 				}
 
-				@Override
-				public void onError(Throwable t) {
+				intermediate.flip();
+				dataBuffer.write(intermediate);
 
-					if (isClosed()) {
+				sink.next(dataBuffer);
 
-						Operators.onErrorDropped(t, sink.currentContext());
-						return;
+				try {
+					if (bytes == -1) {
+						sink.complete();
 					}
-
+				} finally {
 					onReadDone();
-					DataBufferUtils.release(dataBuffer);
-					Operators.onNextDropped(dataBuffer, sink.currentContext());
-					sink.error(t);
 				}
+			}
 
-				@Override
-				public void onComplete() {
+			@Override
+			public void onError(Throwable t) {
 
-					if (onShouldRead()) {
-						emitNext(sink);
-					}
+				if (isClosed()) {
+
+					Operators.onErrorDropped(t, sink.currentContext());
+					return;
 				}
-			});
+
+				onReadDone();
+				DataBufferUtils.release(dataBuffer);
+				Operators.onNextDropped(dataBuffer, sink.currentContext());
+				sink.error(t);
+			}
+
+			@Override
+			public void onComplete() {
+
+				if (onShouldRead()) {
+					emitNext(sink);
+				}
+			}
 		}
 	}
 }