jdk-sandbox: comparison src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/internal/websocket/WebSocketImpl.java

equal deleted inserted replaced

-:dfa9489d1cb1
+:77feac3903d9
 */
 package jdk.incubator.http.internal.websocket;
 import jdk.incubator.http.WebSocket;
+import jdk.incubator.http.internal.common.Demand;
 import jdk.incubator.http.internal.common.Log;
 import jdk.incubator.http.internal.common.MinimalFuture;
 import jdk.incubator.http.internal.common.Pair;
 import jdk.incubator.http.internal.common.SequentialScheduler;
 import jdk.incubator.http.internal.common.SequentialScheduler.DeferredCompleter;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import static java.util.Objects.requireNonNull;
 import static jdk.incubator.http.internal.common.MinimalFuture.failedFuture;
 import static jdk.incubator.http.internal.common.Pair.pair;
 import static jdk.incubator.http.internal.websocket.StatusCodes.CLOSED_ABNORMALLY;
 import static jdk.incubator.http.internal.websocket.StatusCodes.NO_STATUS_CODE;
 import static jdk.incubator.http.internal.websocket.StatusCodes.isLegalToSendFromClient;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.BINARY;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.CLOSE;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.ERROR;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.IDLE;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.OPEN;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.PING;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.PONG;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.TEXT;
+import static jdk.incubator.http.internal.websocket.WebSocketImpl.State.WAITING;
 /*
 * A WebSocket client.
 */
 public final class WebSocketImpl implements WebSocket {
-private static final int IDLE   =  0;
+enum State {
-private static final int OPEN   =  1;
+OPEN,
-private static final int TEXT   =  2;
+IDLE,
-private static final int BINARY =  4;
+WAITING,
-private static final int PING   =  8;
+TEXT,
-private static final int PONG   = 16;
+BINARY,
-private static final int CLOSE  = 32;
+PING,
-private static final int ERROR  = 64;
+PONG,
+CLOSE,
+ERROR;
+}
 private volatile boolean inputClosed;
 private volatile boolean outputClosed;
-/* Which of the listener's methods to call next? */
+private final AtomicReference<State> state = new AtomicReference<>(OPEN);
-private final AtomicInteger state = new AtomicInteger(OPEN);
 /* Components of calls to Listener's methods */
 private MessagePart part;
 private ByteBuffer binaryData;
 private CharSequence text;
 queue = new ConcurrentLinkedQueue<>();
 private final Context context = new OutgoingMessage.Context();
 private final Transmitter transmitter;
 private final Receiver receiver;
 private final SequentialScheduler receiveScheduler = new SequentialScheduler(new ReceiveTask());
+private final Demand demand = new Demand();
 public static CompletableFuture<WebSocket> newInstanceAsync(BuilderImpl b) {
 Function<Result, WebSocket> newWebSocket = r -> {
 WebSocket ws = newInstance(b.getUri(),
 r.subprotocol,
 }
 private WebSocketImpl(URI uri,
 String subprotocol,
 Listener listener,
-TransportSupplier transport)
+TransportSupplier transport) {
-{
 this.uri = requireNonNull(uri);
 this.subprotocol = requireNonNull(subprotocol);
 this.listener = requireNonNull(listener);
 this.transmitter = transport.transmitter();
 this.receiver = transport.receiver(new SignallingMessageConsumer());
 * fashion in respect to other messages accepted through this method. No
 * further messages will be accepted until the returned CompletableFuture
 * completes. This method is used to enforce "one outstanding send
 * operation" policy.
 */
-private CompletableFuture<WebSocket> enqueueExclusively(OutgoingMessage m)
+private CompletableFuture<WebSocket> enqueueExclusively(OutgoingMessage m) {
-{
 if (!outstandingSend.compareAndSet(false, true)) {
 return failedFuture(new IllegalStateException("Send pending"));
 }
 return enqueue(m).whenComplete((r, e) -> outstandingSend.set(false));
 }
 }
 }
 @Override
 public void request(long n) {
-// TODO: delay until state becomes ACTIVE, otherwise messages might be
+if (demand.increase(n)) {
-// requested and consecutively become pending before onOpen is signalled
+receiveScheduler.runOrSchedule();
-receiver.request(n);
+}
 }
 @Override
 public String getSubprotocol() {
 return subprotocol;
 *     - after the state has been observed as CLOSE/ERROR, the scheduler
 *       is stopped
 */
 private class ReceiveTask extends SequentialScheduler.CompleteRestartableTask {
+// Receiver only asked here and nowhere else because we must make sure
+// onOpen is invoked first and no messages become pending before onOpen
+// finishes
 @Override
 public void run() {
-final int s = state.getAndSet(IDLE);
+while (true) {
-try {
+State s = state.get();
-switch (s) {
+try {
-case OPEN:
+switch (s) {
-processOpen();
+case OPEN:
-break;
+processOpen();
-case TEXT:
+tryChangeState(OPEN, IDLE);
-processText();
+break;
-break;
+case TEXT:
-case BINARY:
+processText();
-processBinary();
+tryChangeState(TEXT, IDLE);
 break;
-case PING:
+case BINARY:
-processPing();
+processBinary();
-break;
+tryChangeState(BINARY, IDLE);
-case PONG:
+break;
-processPong();
+case PING:
-break;
+processPing();
-case CLOSE:
+tryChangeState(PING, IDLE);
-processClose();
+break;
-break;
+case PONG:
-case ERROR:
+processPong();
-processError();
+tryChangeState(PONG, IDLE);
 break;
-case IDLE:
+case CLOSE:
-// For debugging spurious signalling: when there was a
+processClose();
-// signal, but apparently nothing has changed
+return;
-break;
+case ERROR:
-default:
+processError();
-throw new InternalError(String.valueOf(s));
+return;
+case IDLE:
+if (demand.tryDecrement()
+&& tryChangeState(IDLE, WAITING)) {
+receiver.request(1);
+}
+return;
+case WAITING:
+// For debugging spurious signalling: when there was a
+// signal, but apparently nothing has changed
+return;
+default:
+throw new InternalError(String.valueOf(s));
+}
+} catch (Throwable t) {
+signalError(t);
 }
-} catch (Throwable t) {
-signalError(t);
 }
 }
 private void processError() throws IOException {
 receiver.close();
 }
 private void signalError(Throwable error) {
 inputClosed = true;
 outputClosed = true;
-if (!this.error.compareAndSet(null, error) || !tryChangeState(ERROR)) {
+if (!this.error.compareAndSet(null, error) || !trySetState(ERROR)) {
 Log.logError(error);
 } else {
 close();
 }
 }
 */
 private void signalClose(int statusCode, String reason) {
 inputClosed = true;
 this.statusCode = statusCode;
 this.reason = reason;
-if (!tryChangeState(CLOSE)) {
+if (!trySetState(CLOSE)) {
 Log.logTrace("Close: {0}, ''{1}''", statusCode, reason);
 } else {
 try {
 receiver.close();
 } catch (Throwable t) {
 @Override
 public void onText(CharSequence data, MessagePart part) {
 receiver.acknowledge();
 text = data;
 WebSocketImpl.this.part = part;
-tryChangeState(TEXT);
+tryChangeState(WAITING, TEXT);
 }
 @Override
 public void onBinary(ByteBuffer data, MessagePart part) {
 receiver.acknowledge();
 binaryData = data;
 WebSocketImpl.this.part = part;
-tryChangeState(BINARY);
+tryChangeState(WAITING, BINARY);
 }
 @Override
 public void onPing(ByteBuffer data) {
 receiver.acknowledge();
 binaryData = data;
-tryChangeState(PING);
+tryChangeState(WAITING, PING);
 }
 @Override
 public void onPong(ByteBuffer data) {
 receiver.acknowledge();
 binaryData = data;
-tryChangeState(PONG);
+tryChangeState(WAITING, PONG);
 }
 @Override
 public void onClose(int statusCode, CharSequence reason) {
 receiver.acknowledge();
 signalClose(statusCode, reason.toString());
 }
 @Override
 public void onComplete() {
+receiver.acknowledge();
 signalClose(CLOSED_ABNORMALLY, "");
 }
 @Override
 public void onError(Throwable error) {
 signalError(error);
 }
 }
-private boolean tryChangeState(int newState) {
+private boolean trySetState(State newState) {
 while (true) {
-int currentState = state.get();
+State currentState = state.get();
 if (currentState == ERROR || currentState == CLOSE) {
 return false;
 } else if (state.compareAndSet(currentState, newState)) {
 receiveScheduler.runOrSchedule();
 return true;
 }
 }
 }
+private boolean tryChangeState(State expectedState, State newState) {
+State witness = state.compareAndExchange(expectedState, newState);
+if (witness == expectedState) {
+receiveScheduler.runOrSchedule();
+return true;
+}
+// This should be the only reason for inability to change the state from
+// IDLE to WAITING: the state has changed to terminal
+if (witness != ERROR && witness != CLOSE) {
+throw new InternalError();
+}
+return false;
+}
 }

branch	http-client-branch
changeset 55922	77feac3903d9
parent 55907	f6a3a657416e
child 55973	4d9b002587db