--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/jdk/src/java.httpclient/share/classes/java/net/http/WSReceiver.java Mon May 09 23:33:09 2016 +0100
@@ -0,0 +1,275 @@
+/*
+ * Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation. Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package java.net.http;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.ProtocolException;
+import java.net.http.WebSocket.Listener;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+
+import static java.lang.System.Logger.Level.ERROR;
+import static java.net.http.WSUtils.EMPTY_BYTE_BUFFER;
+import static java.net.http.WSUtils.logger;
+
+/*
+ * Receives incoming data from the channel and converts it into a sequence of
+ * messages, which are then passed to the listener.
+ */
+final class WSReceiver {
+
+ private final Listener listener;
+ private final WebSocket webSocket;
+ private final Supplier<WSShared<ByteBuffer>> buffersSupplier =
+ new WSSharedPool<>(() -> ByteBuffer.allocateDirect(32768), 2);
+ private final RawChannel channel;
+ private final RawChannel.NonBlockingEvent channelEvent;
+ private final WSSignalHandler handler;
+ private final AtomicLong demand = new AtomicLong();
+ private final AtomicBoolean readable = new AtomicBoolean();
+ private boolean started;
+ private volatile boolean closed;
+ private final WSFrame.Reader reader = new WSFrame.Reader();
+ private final WSFrameConsumer frameConsumer;
+ private WSShared<ByteBuffer> buf = WSShared.wrap(EMPTY_BYTE_BUFFER);
+ private WSShared<ByteBuffer> data; // TODO: initialize with leftovers from the RawChannel
+
+ WSReceiver(Listener listener, WebSocket webSocket, Executor executor,
+ RawChannel channel) {
+ this.listener = listener;
+ this.webSocket = webSocket;
+ this.channel = channel;
+ handler = new WSSignalHandler(executor, this::react);
+ channelEvent = createChannelEvent();
+ this.frameConsumer = new WSFrameConsumer(new MessageConsumer());
+ }
+
+ private void react() {
+ synchronized (this) {
+ while (demand.get() > 0 && !closed) {
+ try {
+ if (data == null) {
+ if (!getData()) {
+ break;
+ }
+ }
+ reader.readFrame(data, frameConsumer);
+ if (!data.hasRemaining()) {
+ data.dispose();
+ data = null;
+ }
+ // In case of exception we don't need to clean any state,
+ // since it's the terminal condition anyway. Nothing will be
+ // retried.
+ } catch (WSProtocolException e) {
+ // Translate into ProtocolException
+ closeExceptionally(new ProtocolException().initCause(e));
+ } catch (Exception e) {
+ closeExceptionally(e);
+ }
+ }
+ }
+ }
+
+ long request(long n) {
+ long newDemand = demand.accumulateAndGet(n, (p, i) -> p + i < 0 ? Long.MAX_VALUE : p + i);
+ handler.signal();
+ assert newDemand >= 0 : newDemand;
+ return newDemand;
+ }
+
+ private boolean getData() throws IOException {
+ if (!readable.get()) {
+ return false;
+ }
+ if (!buf.hasRemaining()) {
+ buf.dispose();
+ buf = buffersSupplier.get();
+ assert buf.hasRemaining() : buf;
+ }
+ int oldPosition = buf.position();
+ int oldLimit = buf.limit();
+ int numRead = channel.read(buf.buffer());
+ if (numRead > 0) {
+ data = buf.share(oldPosition, oldPosition + numRead);
+ buf.select(buf.limit(), oldLimit); // Move window to the free region
+ return true;
+ } else if (numRead == 0) {
+ readable.set(false);
+ channel.registerEvent(channelEvent);
+ return false;
+ } else {
+ assert numRead < 0 : numRead;
+ throw new WSProtocolException
+ ("7.2.1.", "Stream ended before a Close frame has been received");
+ }
+ }
+
+ void start() {
+ synchronized (this) {
+ if (started) {
+ throw new IllegalStateException("Already started");
+ }
+ started = true;
+ try {
+ channel.registerEvent(channelEvent);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ try {
+ listener.onOpen(webSocket);
+ } catch (Exception e) {
+ closeExceptionally(new RuntimeException("onOpen threw an exception", e));
+ }
+ }
+ }
+
+ private void close() { // TODO: move to WS.java
+ closed = true;
+ }
+
+ private void closeExceptionally(Throwable error) { // TODO: move to WS.java
+ close();
+ try {
+ listener.onError(webSocket, error);
+ } catch (Exception e) {
+ logger.log(ERROR, "onError threw an exception", e);
+ }
+ }
+
+ private final class MessageConsumer implements WSMessageConsumer {
+
+ @Override
+ public void onText(WebSocket.MessagePart part, WSDisposableText data) {
+ decrementDemand();
+ CompletionStage<?> cs;
+ try {
+ cs = listener.onText(webSocket, data, part);
+ } catch (Exception e) {
+ closeExceptionally(new RuntimeException("onText threw an exception", e));
+ return;
+ }
+ follow(cs, data, "onText");
+ }
+
+ @Override
+ public void onBinary(WebSocket.MessagePart part, WSShared<ByteBuffer> data) {
+ decrementDemand();
+ CompletionStage<?> cs;
+ try {
+ cs = listener.onBinary(webSocket, data.buffer(), part);
+ } catch (Exception e) {
+ closeExceptionally(new RuntimeException("onBinary threw an exception", e));
+ return;
+ }
+ follow(cs, data, "onBinary");
+ }
+
+ @Override
+ public void onPing(WSShared<ByteBuffer> data) {
+ decrementDemand();
+ CompletionStage<?> cs;
+ try {
+ cs = listener.onPing(webSocket, data.buffer());
+ } catch (Exception e) {
+ closeExceptionally(new RuntimeException("onPing threw an exception", e));
+ return;
+ }
+ follow(cs, data, "onPing");
+ }
+
+ @Override
+ public void onPong(WSShared<ByteBuffer> data) {
+ decrementDemand();
+ CompletionStage<?> cs;
+ try {
+ cs = listener.onPong(webSocket, data.buffer());
+ } catch (Exception e) {
+ closeExceptionally(new RuntimeException("onPong threw an exception", e));
+ return;
+ }
+ follow(cs, data, "onPong");
+ }
+
+ @Override
+ public void onClose(WebSocket.CloseCode code, CharSequence reason) {
+ decrementDemand();
+ try {
+ close();
+ listener.onClose(webSocket, Optional.ofNullable(code), reason.toString());
+ } catch (Exception e) {
+ logger.log(ERROR, "onClose threw an exception", e);
+ }
+ }
+ }
+
+ private void follow(CompletionStage<?> cs, WSDisposable d, String source) {
+ if (cs == null) {
+ d.dispose();
+ } else {
+ cs.whenComplete((whatever, error) -> {
+ if (error != null) {
+ String m = "CompletionStage returned by " + source + " completed exceptionally";
+ closeExceptionally(new RuntimeException(m, error));
+ }
+ d.dispose();
+ });
+ }
+ }
+
+ private void decrementDemand() {
+ long newDemand = demand.decrementAndGet();
+ assert newDemand >= 0 : newDemand;
+ }
+
+ private RawChannel.NonBlockingEvent createChannelEvent() {
+ return new RawChannel.NonBlockingEvent() {
+
+ @Override
+ public int interestOps() {
+ return SelectionKey.OP_READ;
+ }
+
+ @Override
+ public void handle() {
+ boolean wasNotReadable = readable.compareAndSet(false, true);
+ assert wasNotReadable;
+ handler.signal();
+ }
+
+ @Override
+ public String toString() {
+ return "Read readiness event [" + channel + "]";
+ }
+ };
+ }
+}