jdk-sandbox: comparison src/java.net.http/share/classes/jdk/internal/net/http/SocketTube.java

equal deleted inserted replaced

-:98075dcd4ffc
+:e453d21310bd
 * questions.
 */
 package jdk.internal.net.http;
-import java.io.EOFException;
 import java.io.IOException;
 import java.lang.System.Logger.Level;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Objects;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import jdk.internal.net.http.common.Demand;
 import jdk.internal.net.http.common.FlowTube;
 import jdk.internal.net.http.common.SequentialScheduler;
 import jdk.internal.net.http.common.SequentialScheduler.DeferredCompleter;
 import jdk.internal.net.http.common.SequentialScheduler.RestartableTask;
 import jdk.internal.net.http.common.Utils;
 /**
 * A SocketTube is a terminal tube plugged directly into the socket.
 * The read subscriber should call {@code subscribe} on the SocketTube before
-* the SocketTube can be subscribed to the write publisher.
+* the SocketTube is subscribed to the write publisher.
 */
 final class SocketTube implements FlowTube {
 static final boolean DEBUG = Utils.DEBUG; // revisit: temporary developer's flag
 final System.Logger  debug = Utils.getDebugLogger(this::dbgString, DEBUG);
 // ===================================================================== //
 //                           Events                                      //
 // ======================================================================//
 void signalClosed() {
-// Ensure that the subscriber will be terminated
+// Ensures that the subscriber will be terminated and that future
-// and that future subscribers will be notified
+// subscribers will be notified when the connection is closed.
-// when the connection is closed.
 readPublisher.subscriptionImpl.signalError(
 new IOException("connection closed locally"));
 }
 /**
 taskCompleter.complete();
 }
 }
 }
-// This is best effort - there's no guarantee that the printed set
+// This is best effort - there's no guarantee that the printed set of values
-// of values is consistent. It should only be considered as
+// is consistent. It should only be considered as weakly accurate - in
-// weakly accurate - in particular in what concerns the events states,
+// particular in what concerns the events states, especially when displaying
-// especially when displaying a read event state from a write event
+// a read event state from a write event callback and conversely.
-// callback and conversely.
 void debugState(String when) {
 if (debug.isLoggable(Level.DEBUG)) {
 StringBuilder state = new StringBuilder();
 InternalReadPublisher.InternalReadSubscription sub =
 debug.log(Level.DEBUG, state.toString());
 }
 }
 /**
-* A repeatable event that can be paused or resumed by changing
+* A repeatable event that can be paused or resumed by changing its
-* its interestOps.
+* interestOps. When the event is fired, it is first paused before being
-* When the event is fired, it is first paused before being signaled.
+* signaled. It is the responsibility of the code triggered by
-* It is the responsibility of the code triggered by {@code signalEvent}
+* {@code signalEvent} to resume the event if required.
-* to resume the event if required.
 */
 private static abstract class SocketFlowEvent extends AsyncEvent {
 final SocketChannel channel;
 final int defaultInterest;
 volatile int interestOps;
 // ===================================================================== //
 //                              Writing                                  //
 // ======================================================================//
-// This class makes the assumption that the publisher will call
+// This class makes the assumption that the publisher will call onNext
-// onNext sequentially, and that onNext won't be called if the demand
+// sequentially, and that onNext won't be called if the demand has not been
-// has not been incremented by request(1).
+// incremented by request(1).
 // It has a 'queue of 1' meaning that it will call request(1) in
 // onSubscribe, and then only after its 'current' buffer list has been
 // fully written and current set to null;
 private final class InternalWriteSubscriber
 implements Flow.Subscriber<List<ByteBuffer>> {
 // ReadEvent -> rs.onNext(bytes) -> process server SETTINGS -> write
 // client SETTINGS -> ws.onNext(bytes) -> tryFlushCurrent
 debugState("leaving w.onNext");
 }
-// we don't use a SequentialScheduler here: we rely on
+// Don't use a SequentialScheduler here: rely on onNext() being invoked
-// onNext() being called sequentially, and not being called
+// sequentially, and not being invoked if there is no demand, request(1).
-// if we haven't call request(1)
+// onNext is usually called from within a user / executor thread.
-// onNext is usually called from within a user/executor thread.
+// Initial writing will be performed in that thread. If for some reason,
-// we will perform the initial writing in that thread.
+// not all the data can be written, a writeEvent will be registered, and
-// if for some reason, not all data can be written, the writeEvent
+// writing will resume in the the selector manager thread when the
-// will be resumed, and the rest of the data will be written from
+// writeEvent is fired.
-// the selector manager thread when the writeEvent is fired.
+//
-// If we are in the selector manager thread, then we will use the executor
+// If this method is invoked in the selector manager thread (because of
-// to call request(1), ensuring that onNext() won't be called from
+// a writeEvent), then the executor will be used to invoke request(1),
-// within the selector thread.
+// ensuring that onNext() won't be invoked from within the selector
-// If we are not in the selector manager thread, then we don't care.
+// thread. If not in the selector manager thread, then request(1) is
+// invoked directly.
 void tryFlushCurrent(boolean inSelectorThread) {
 List<ByteBuffer> bufs = current;
 if (bufs == null) return;
 try {
 assert inSelectorThread == client.isSelectorThread() :
 + " be in the selector thread";
 long remaining = Utils.remaining(bufs);
 debug.log(Level.DEBUG, "trying to write: %d", remaining);
 long written = writeAvailable(bufs);
 debug.log(Level.DEBUG, "wrote: %d", written);
-if (written == -1) {
+assert written >= 0 : "negative number of bytes written:" + written;
-signalError(new EOFException("EOF reached while writing"));
-return;
-}
 assert written <= remaining;
 if (remaining - written == 0) {
 current = null;
 if (writeDemand.tryDecrement()) {
 Runnable requestMore = this::requestMore;
 signalError(t);
 subscription.cancel();
 }
 }
-// Kick off the initial request:1 that will start
+// Kick off the initial request:1 that will start the writing side.
-// the writing side. Called from the selector manager
+// Invoked in the selector manager thread.
-// thread.
 void startSubscription() {
 try {
 debug.log(Level.DEBUG, "write: starting subscription");
 assert client.isSelectorThread();
 // make sure read registrations are handled before;
 final long remaining = Utils.remaining(srcs);
 long written = 0;
 while (remaining > written) {
 try {
 long w = channel.write(srcs);
-if (w == -1 && written == 0) return -1;
+assert w >= 0 : "negative number of bytes written:" + w;
-if (w == 0) break;
+if (w == 0) {
+break;
+}
 written += w;
 } catch (IOException x) {
-// if no bytes were written just throws...
+if (written == 0) {
-if (written == 0) throw x;
+// no bytes were written just throw
-// otherwise return how many bytes were
+throw x;
-// written: we will fail next time.
+} else {
-break;
+// return how many bytes were written, will fail next time
+break;
+}
 }
 }
 return written;
 }

branch	http-client-branch
changeset 56374	e453d21310bd
parent 56372	98075dcd4ffc
child 56377	eef94a3576a4