Compose Request(n) – Transport/Application

src/main/java/io/reactivesocket/internal/Responder.java

@@ -118,7 +118,7 @@ private void start(final Completable responderCompletable) {
 		/* state of cancellation subjects during connection */
 		final Int2ObjectHashMap<Subscription> cancellationSubscriptions = new Int2ObjectHashMap<>();
 		/* streams in flight that can receive REQUEST_N messages */
-		final Int2ObjectHashMap<Subscription> inFlight = new Int2ObjectHashMap<>(); // TODO not being used
+		final Int2ObjectHashMap<SubscriptionArbiter> inFlight = new Int2ObjectHashMap<>(); 
 		/* bidirectional channels */
 		final Int2ObjectHashMap<UnicastSubject<Payload>> channels = new Int2ObjectHashMap<>(); // TODO should/can we make this optional so that it only gets allocated per connection if channels are
 																									// used?
@@ -205,14 +205,14 @@ public void onNext(Frame requestFrame) {
 							}
 							return;
 						} else if (requestFrame.getType() == FrameType.REQUEST_N) {
-							Subscription inFlightSubscription = null;
+							SubscriptionArbiter inFlightSubscription = null;
 							synchronized (Responder.this)
 							{
 								inFlightSubscription = inFlight.get(requestFrame.getStreamId());
 							}
 							if (inFlightSubscription != null)
 							{
-								inFlightSubscription.request(Frame.RequestN.requestN(requestFrame));
+								inFlightSubscription.addApplicationRequest(Frame.RequestN.requestN(requestFrame));
 								return;
 							}
 							// TODO should we do anything if we don't find the stream? emitting an error is risky as the responder could have terminated and cleaned up already
@@ -407,15 +407,15 @@ private Publisher<Frame> handleRequestStream(
 			Frame requestFrame,
 			final RequestHandler requestHandler,
 			final Int2ObjectHashMap<Subscription> cancellationSubscriptions,
-			final Int2ObjectHashMap<Subscription> inFlight) {
+			final Int2ObjectHashMap<SubscriptionArbiter> inFlight) {
 		return _handleRequestStream(requestStreamHandler, requestFrame, requestHandler, cancellationSubscriptions, inFlight, true);
 	}
 
 	private Publisher<Frame> handleRequestSubscription(
 			Frame requestFrame,
 			final RequestHandler requestHandler,
 			final Int2ObjectHashMap<Subscription> cancellationSubscriptions,
-			final Int2ObjectHashMap<Subscription> inFlight) {
+			final Int2ObjectHashMap<SubscriptionArbiter> inFlight) {
 		return _handleRequestStream(requestSubscriptionHandler, requestFrame, requestHandler, cancellationSubscriptions, inFlight, false);
 	}
 
@@ -434,7 +434,7 @@ private Publisher<Frame> _handleRequestStream(
 			Frame requestFrame,
 			final RequestHandler requestHandler,
 			final Int2ObjectHashMap<Subscription> cancellationSubscriptions,
-			final Int2ObjectHashMap<Subscription> inFlight,
+			final Int2ObjectHashMap<SubscriptionArbiter> inFlight,
 			final boolean allowCompletion) {
 
 		return new Publisher<Frame>() {
@@ -445,19 +445,25 @@ public void subscribe(Subscriber<? super Frame> child) {
 
 					final AtomicBoolean started = new AtomicBoolean(false);
 					final AtomicReference<Subscription> parent = new AtomicReference<>();
+					final SubscriptionArbiter arbiter = new SubscriptionArbiter();
 
 					@Override
 					public void request(long n) {
-						if (n > 0 && started.compareAndSet(false,  true)) {
+						if(n <= 0) {
+							return;
+						}
+						if (started.compareAndSet(false,  true)) {
+							arbiter.addTransportRequest(n);
 							final int streamId = requestFrame.getStreamId();
 
 							handler.apply(requestHandler, requestFrame).subscribe(new Subscriber<Payload>() {
 
 								@Override
 								public void onSubscribe(Subscription s) {
 									if (parent.compareAndSet(null, s)) {
-										inFlight.put(streamId, s);
-										s.request(Frame.Request.initialRequestN(requestFrame));
+										inFlight.put(streamId, arbiter);
+										arbiter.addApplicationRequest(Frame.Request.initialRequestN(requestFrame));
+										arbiter.addApplicationProducer(s);
 									} else {
 										s.cancel();
 										cleanup();
@@ -493,6 +499,8 @@ public void onComplete() {
 								}
 
 							});
+						} else {
+							arbiter.addTransportRequest(n);
 						}
 					}
 
@@ -561,7 +569,7 @@ private Publisher<Frame> handleRequestChannel(Frame requestFrame,
 			RequestHandler requestHandler,
 			Int2ObjectHashMap<UnicastSubject<Payload>> channels,
 			Int2ObjectHashMap<Subscription> cancellationSubscriptions,
-			Int2ObjectHashMap<Subscription> inFlight) {
+			Int2ObjectHashMap<SubscriptionArbiter> inFlight) {
 
 		UnicastSubject<Payload> channelSubject = null;
 		synchronized(Responder.this) {
@@ -576,10 +584,15 @@ public void subscribe(Subscriber<? super Frame> child) {
 
 						final AtomicBoolean started = new AtomicBoolean(false);
 						final AtomicReference<Subscription> parent = new AtomicReference<>();
+						final SubscriptionArbiter arbiter = new SubscriptionArbiter();
 
 						@Override
 						public void request(long n) {
-							if (n > 0 && started.compareAndSet(false, true)) {
+							if(n <= 0) {
+								return;
+							}
+							if (started.compareAndSet(false, true)) {
+								arbiter.addTransportRequest(n);
 								final int streamId = requestFrame.getStreamId();
 
 								// first request on this channel
@@ -609,8 +622,9 @@ public void request(long n) {
 									@Override
 									public void onSubscribe(Subscription s) {
 										if (parent.compareAndSet(null, s)) {
-											inFlight.put(streamId, s);
-											s.request(Frame.Request.initialRequestN(requestFrame));
+											inFlight.put(streamId, arbiter);
+											arbiter.addApplicationRequest(Frame.Request.initialRequestN(requestFrame));
+											arbiter.addApplicationProducer(s);
 										} else {
 											s.cancel();
 											cleanup();
@@ -638,6 +652,8 @@ public void onComplete() {
 									}
 
 								});
+							} else {
+								arbiter.addTransportRequest(n);
 							}
 						}
 
@@ -681,5 +697,51 @@ private void cleanup() {
 			}
 		}
 	}
+
+	private static class SubscriptionArbiter {
+		private Subscription applicationProducer;
+		private long appRequested = 0;
+		private long transportRequested = 0;
+		private long requestedToProducer = 0;
+
+		public void addApplicationRequest(long n) {
+			synchronized(this) {
+				appRequested += n;
+			}
+			tryRequest();
+		}
+
+		public void addApplicationProducer(Subscription s) {
+			synchronized(this) {
+				applicationProducer = s;
+			}
+			tryRequest();
+		}
+
+		public void addTransportRequest(long n) {
+			synchronized(this) {
+				transportRequested += n;
+			}
+			tryRequest();
+		}
+
+		private void tryRequest() {
+			long toRequest = 0;
+			Subscription s = null;
+			synchronized(this) {
+				if(applicationProducer == null) {
+					return;
+				}
+				s = applicationProducer;
+				long minToRequest = Math.min(appRequested, transportRequested);
+				toRequest = minToRequest - requestedToProducer;
+				requestedToProducer += toRequest;
+			}
+			if(toRequest > 0) {
+				s.request(toRequest);
+			}
+		}
+
+	}
 
 }

src/test/java/io/reactivesocket/SerializedEventBus.java

@@ -0,0 +1,80 @@
+/**
+ * Copyright 2015 Netflix, Inc.
+ * 
+ * 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 io.reactivesocket;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+import io.reactivesocket.observable.Observer;
+import io.reactivex.subjects.PublishSubject;
+import io.reactivex.subjects.Subject;
+
+/**
+ * Multicast eventbus that serializes incoming events.
+ */
+public class SerializedEventBus {
+
+	private final CopyOnWriteArrayList<Observer<Frame>> os = new CopyOnWriteArrayList<>();
+	private Subject<Frame, Frame> s;
+
+	public SerializedEventBus() {
+		s = PublishSubject.<Frame>create().toSerialized();
+		s.subscribe(f-> {
+			for (Observer<Frame> o : os) {
+				o.onNext(f);
+			}	
+		});
+	}
+
+	public void send(Frame f) {
+		s.onNext(f);
+	}
+
+	public void add(Observer<Frame> o) {
+		os.add(o);
+	}
+
+	public void add(Consumer<Frame> f) {
+		add(new Observer<Frame>() {
+
+			@Override
+			public void onNext(Frame t) {
+				f.accept(t);
+			}
+
+			@Override
+			public void onError(Throwable e) {
+
+			}
+
+			@Override
+			public void onComplete() {
+
+			}
+
+			@Override
+			public void onSubscribe(io.reactivesocket.observable.Disposable d) {
+				// TODO Auto-generated method stub
+
+			}
+
+		});
+	}
+
+	public void remove(Observer<Frame> o) {
+		os.remove(o);
+	}
+}

src/test/java/io/reactivesocket/TestConnection.java

@@ -18,22 +18,18 @@
 import static io.reactivex.Observable.*;
 
 import java.io.IOException;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.function.Consumer;
 
 import org.reactivestreams.Publisher;
 
 import io.reactivesocket.observable.Observer;
 import io.reactivex.Observable;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subjects.PublishSubject;
-import io.reactivex.subjects.Subject;
 
 public class TestConnection implements DuplexConnection {
 
-	public final Channel toInput = new Channel();
-	public final Channel write = new Channel();
+	public final SerializedEventBus toInput = new SerializedEventBus();
+	public final SerializedEventBus write = new SerializedEventBus();
 
 	@Override
 	public void addOutput(Publisher<Frame> o, Completable callback) {
@@ -70,6 +66,9 @@ public void connectToServerConnection(TestConnection serverConnection) {
 		connectToServerConnection(serverConnection, true);
 	}
 
+	Worker clientThread = Schedulers.newThread().createWorker();
+	Worker serverThread = Schedulers.newThread().createWorker();
+
 	public void connectToServerConnection(TestConnection serverConnection, boolean log) {
 		if (log) {
 			serverConnection.write.add(n -> System.out.println("SERVER ==> Writes from server->client: " + n + "   Written from " + Thread.currentThread()));
@@ -78,9 +77,6 @@ public void connectToServerConnection(TestConnection serverConnection, boolean l
 			toInput.add(n -> System.out.println("CLIENT <== Input from server->client: " + n + "   Read on " + Thread.currentThread()));
 		}
 
-		Worker clientThread = Schedulers.newThread().createWorker();
-		Worker serverThread = Schedulers.newThread().createWorker();
-
 		// client to server
 		write.add(f -> {
 			serverThread.schedule(() -> {
@@ -97,61 +93,8 @@ public void connectToServerConnection(TestConnection serverConnection, boolean l
 
 	@Override
 	public void close() throws IOException {
-
-	}
-
-	public static class Channel {
-
-		private final CopyOnWriteArrayList<Observer<Frame>> os = new CopyOnWriteArrayList<>();
-		private Subject<Frame, Frame> s;
-
-		public Channel() {
-			s = PublishSubject.<Frame>create().toSerialized();
-			s.subscribe(f-> {
-				for (Observer<Frame> o : os) {
-					o.onNext(f);
-				}	
-			});
-		}
-
-		public void send(Frame f) {
-			s.onNext(f);
-		}
-
-		public void add(Observer<Frame> o) {
-			os.add(o);
-		}
-
-		public void add(Consumer<Frame> f) {
-			add(new Observer<Frame>() {
-
-				@Override
-				public void onNext(Frame t) {
-					f.accept(t);
-				}
-
-				@Override
-				public void onError(Throwable e) {
-
-				}
-
-				@Override
-				public void onComplete() {
-
-				}
-
-				@Override
-				public void onSubscribe(io.reactivesocket.observable.Disposable d) {
-					// TODO Auto-generated method stub
-
-				}
-
-			});
-		}
-
-		public void remove(Observer<Frame> o) {
-			os.remove(o);
-		}
+		clientThread.dispose();
+		serverThread.dispose();
 	}
 
 }