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

equal deleted inserted replaced

-:818a23db260c
+:4254bed3c09d
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import jdk.internal.net.http.common.BufferSupplier;
 import jdk.internal.net.http.common.Demand;
 import jdk.internal.net.http.common.FlowTube;
+import jdk.internal.net.http.common.Log;
 import jdk.internal.net.http.common.Logger;
 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;
 // ======================================================================//
 void signalClosed() {
 // Ensures that the subscriber will be terminated and that future
 // subscribers will be notified when the connection is closed.
+if (Log.channel()) {
+Log.logChannel("Connection close signalled: connection closed locally ({0})",
+channelDescr());
+}
 readPublisher.subscriptionImpl.signalError(
 new IOException("connection closed locally"));
 }
 /**
 // Kick off the initial request:1 that will start the writing side.
 // Invoked in the selector manager thread.
 void startSubscription() {
 try {
 if (debug.on()) debug.log("write: starting subscription");
+if (Log.channel()) {
+Log.logChannel("Start requesting bytes for writing to channel: {0}",
+channelDescr());
+}
 assert client.isSelectorThread();
 // make sure read registrations are handled before;
 readPublisher.subscriptionImpl.handlePending();
 if (debug.on()) debug.log("write: offloading requestMore");
 // start writing;
 tryFlushCurrent(true);
 }
 void signalError(Throwable error) {
 debug.log(() -> "write error: " + error);
+if (Log.channel()) {
+Log.logChannel("Failed to write to channel ({0}: {1})",
+channelDescr(), error);
+}
 completed = true;
 readPublisher.signalError(error);
 }
 // A repeatable WriteEvent which is paused after firing and can
 upstreamSubscription.request(n);
 }
 @Override
 public void cancel() {
+if (debug.on()) debug.log("write: cancel");
 dropSubscription();
 upstreamSubscription.cancel();
 }
 void dropSubscription() {
 void signalError(Throwable error) {
 if (debug.on()) debug.log("error signalled " + error);
 if (!errorRef.compareAndSet(null, error)) {
 return;
 }
+if (Log.channel()) {
+Log.logChannel("Error signalled on channel {0}: {1}",
+channelDescr(), error);
+}
 subscriptionImpl.handleError();
 }
 final class ReadSubscription implements Flow.Subscription {
 final InternalReadSubscription impl;
 }
 final void handleSubscribeEvent() {
 assert client.isSelectorThread();
 debug.log("subscribe event raised");
+if (Log.channel()) Log.logChannel("Start reading from {0}", channelDescr());
 readScheduler.runOrSchedule();
 if (readScheduler.isStopped() || completed) {
 // if already completed or stopped we can handle any
 // pending connection directly from here.
 if (debug.on())
 }
 @Override
 public final void cancel() {
 pauseReadEvent();
+if (Log.channel()) {
+Log.logChannel("Read subscription cancelled for channel {0}",
+channelDescr());
+}
 readScheduler.stop();
 }
 private void resumeReadEvent() {
 if (debug.on()) debug.log("resuming read event");
 final void signalError(Throwable error) {
 if (!errorRef.compareAndSet(null, error)) {
 return;
 }
 if (debug.on()) debug.log("got read error: " + error);
+if (Log.channel()) {
+Log.logChannel("Read error signalled on channel {0}: {1}",
+channelDescr(), error);
+}
 readScheduler.runOrSchedule();
 }
 final void signalReadable() {
 readScheduler.runOrSchedule();
 // safe to pause here because we're finished anyway.
 pauseReadEvent();
 if (debug.on())
 debug.log("Sending error " + error
 + " to subscriber " + subscriber);
+if (Log.channel()) {
+Log.logChannel("Raising error with subscriber for {0}: {1}",
+channelDescr(), error);
+}
 current.errorRef.compareAndSet(null, error);
 current.signalCompletion();
 readScheduler.stop();
 debugState("leaving read() loop with error: ");
 return;
 try {
 List<ByteBuffer> bytes = readAvailable(current.bufferSource);
 if (bytes == EOF) {
 if (!completed) {
 if (debug.on()) debug.log("got read EOF");
+if (Log.channel()) {
+Log.logChannel("EOF read from channel: {0}",
+channelDescr());
+}
 completed = true;
 // safe to pause here because we're finished
 // anyway.
 pauseReadEvent();
 current.signalCompletion();
 }
 } catch (Throwable t) {
 if (debug.on()) debug.log("Unexpected exception in read loop", t);
 signalError(t);
 } finally {
+if (readScheduler.isStopped()) {
+if (debug.on()) debug.log("Read scheduler stopped");
+if (Log.channel()) {
+Log.logChannel("Stopped reading from channel {0}", channelDescr());
+}
+}
 handlePending();
 }
 }
 boolean handlePending() {
 }
 final String dbgString() {
 return "SocketTube("+id+")";
 }
+final String channelDescr() {
+return String.valueOf(channel);
+}
 }

changeset 50681	4254bed3c09d
parent 49944	4690a2871b44
child 50985	cd41f34e548c
child 56795	03ece2518428