1 /*

     2  * Copyright (c) 2015, 2016, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package java.net.http;

    27 

    28 import java.io.IOException;

    29 import java.io.UncheckedIOException;

    30 import java.net.ProtocolException;

    31 import java.net.http.WebSocket.Listener;

    32 import java.nio.ByteBuffer;

    33 import java.nio.CharBuffer;

    34 import java.nio.channels.SelectionKey;

    35 import java.util.Optional;

    36 import java.util.concurrent.CompletionStage;

    37 import java.util.concurrent.Executor;

    38 import java.util.concurrent.atomic.AtomicBoolean;

    39 import java.util.concurrent.atomic.AtomicLong;

    40 import java.util.function.Supplier;

    41 

    42 import static java.lang.System.Logger.Level.ERROR;

    43 import static java.net.http.WSUtils.EMPTY_BYTE_BUFFER;

    44 import static java.net.http.WSUtils.logger;

    45 

    46 /*

    47  * Receives incoming data from the channel and converts it into a sequence of

    48  * messages, which are then passed to the listener.

    49  */

    50 final class WSReceiver {

    51 

    52     private final Listener listener;

    53     private final WebSocket webSocket;

    54     private final Supplier<WSShared<ByteBuffer>> buffersSupplier =

    55             new WSSharedPool<>(() -> ByteBuffer.allocateDirect(32768), 2);

    56     private final RawChannel channel;

    57     private final RawChannel.RawEvent channelEvent;

    58     private final WSSignalHandler handler;

    59     private final AtomicLong demand = new AtomicLong();

    60     private final AtomicBoolean readable = new AtomicBoolean();

    61     private boolean started;

    62     private volatile boolean closed;

    63     private final WSFrame.Reader reader = new WSFrame.Reader();

    64     private final WSFrameConsumer frameConsumer;

    65     private WSShared<ByteBuffer> buf = WSShared.wrap(EMPTY_BYTE_BUFFER);

    66     private WSShared<ByteBuffer> data; // TODO: initialize with leftovers from the RawChannel

    67 

    68     WSReceiver(Listener listener, WebSocket webSocket, Executor executor,

    69                RawChannel channel) {

    70         this.listener = listener;

    71         this.webSocket = webSocket;

    72         this.channel = channel;

    73         handler = new WSSignalHandler(executor, this::react);

    74         channelEvent = createChannelEvent();

    75         this.frameConsumer = new WSFrameConsumer(new MessageConsumer());

    76     }

    77 

    78     private void react() {

    79         synchronized (this) {

    80             while (demand.get() > 0 && !closed) {

    81                 try {

    82                     if (data == null) {

    83                         if (!getData()) {

    84                             break;

    85                         }

    86                     }

    87                     reader.readFrame(data, frameConsumer);

    88                     if (!data.hasRemaining()) {

    89                         data.dispose();

    90                         data = null;

    91                     }

    92                     // In case of exception we don't need to clean any state,

    93                     // since it's the terminal condition anyway. Nothing will be

    94                     // retried.

    95                 } catch (WSProtocolException e) {

    96                     // Translate into ProtocolException

    97                     closeExceptionally(new ProtocolException().initCause(e));

    98                 } catch (Exception e) {

    99                     closeExceptionally(e);

   100                 }

   101             }

   102         }

   103     }

   104 

   105     void request(long n) {

   106         long newDemand = demand.accumulateAndGet(n, (p, i) -> p + i < 0 ? Long.MAX_VALUE : p + i);

   107         handler.signal();

   108         assert newDemand >= 0 : newDemand;

   109     }

   110 

   111     private boolean getData() throws IOException {

   112         if (!readable.get()) {

   113             return false;

   114         }

   115         if (!buf.hasRemaining()) {

   116             buf.dispose();

   117             buf = buffersSupplier.get();

   118             assert buf.hasRemaining() : buf;

   119         }

   120         int oldPosition = buf.position();

   121         int oldLimit = buf.limit();

   122         int numRead = channel.read(buf.buffer());

   123         if (numRead > 0) {

   124             data = buf.share(oldPosition, oldPosition + numRead);

   125             buf.select(buf.limit(), oldLimit); // Move window to the free region

   126             return true;

   127         } else if (numRead == 0) {

   128             readable.set(false);

   129             channel.registerEvent(channelEvent);

   130             return false;

   131         } else {

   132             assert numRead < 0 : numRead;

   133             throw new WSProtocolException

   134                     ("7.2.1.", "Stream ended before a Close frame has been received");

   135         }

   136     }

   137 

   138     void start() {

   139         synchronized (this) {

   140             if (started) {

   141                 throw new IllegalStateException("Already started");

   142             }

   143             started = true;

   144             try {

   145                 channel.registerEvent(channelEvent);

   146             } catch (IOException e) {

   147                 throw new UncheckedIOException(e);

   148             }

   149             try {

   150                 listener.onOpen(webSocket);

   151             } catch (Exception e) {

   152                 closeExceptionally(new RuntimeException("onOpen threw an exception", e));

   153             }

   154         }

   155     }

   156 

   157     private void close() { // TODO: move to WS.java

   158         closed = true;

   159     }

   160 

   161     private void closeExceptionally(Throwable error) {  // TODO: move to WS.java

   162         close();

   163         try {

   164             listener.onError(webSocket, error);

   165         } catch (Exception e) {

   166             logger.log(ERROR, "onError threw an exception", e);

   167         }

   168     }

   169 

   170     private final class MessageConsumer implements WSMessageConsumer {

   171 

   172         @Override

   173         public void onText(WebSocket.MessagePart part, WSShared<CharBuffer> data) {

   174             decrementDemand();

   175             CompletionStage<?> cs;

   176             try {

   177                 cs = listener.onText(webSocket, data.buffer(), part);

   178             } catch (Exception e) {

   179                 closeExceptionally(new RuntimeException("onText threw an exception", e));

   180                 return;

   181             }

   182             follow(cs, data, "onText");

   183         }

   184 

   185         @Override

   186         public void onBinary(WebSocket.MessagePart part, WSShared<ByteBuffer> data) {

   187             decrementDemand();

   188             CompletionStage<?> cs;

   189             try {

   190                 cs = listener.onBinary(webSocket, data.buffer(), part);

   191             } catch (Exception e) {

   192                 closeExceptionally(new RuntimeException("onBinary threw an exception", e));

   193                 return;

   194             }

   195             follow(cs, data, "onBinary");

   196         }

   197 

   198         @Override

   199         public void onPing(WSShared<ByteBuffer> data) {

   200             decrementDemand();

   201             CompletionStage<?> cs;

   202             try {

   203                 cs = listener.onPing(webSocket, data.buffer());

   204             } catch (Exception e) {

   205                 closeExceptionally(new RuntimeException("onPing threw an exception", e));

   206                 return;

   207             }

   208             follow(cs, data, "onPing");

   209         }

   210 

   211         @Override

   212         public void onPong(WSShared<ByteBuffer> data) {

   213             decrementDemand();

   214             CompletionStage<?> cs;

   215             try {

   216                 cs = listener.onPong(webSocket, data.buffer());

   217             } catch (Exception e) {

   218                 closeExceptionally(new RuntimeException("onPong threw an exception", e));

   219                 return;

   220             }

   221             follow(cs, data, "onPong");

   222         }

   223 

   224         @Override

   225         public void onClose(WebSocket.CloseCode code, CharSequence reason) {

   226             decrementDemand();

   227             try {

   228                 close();

   229                 listener.onClose(webSocket, Optional.ofNullable(code), reason.toString());

   230             } catch (Exception e) {

   231                 logger.log(ERROR, "onClose threw an exception", e);

   232             }

   233         }

   234     }

   235 

   236     private void follow(CompletionStage<?> cs, WSDisposable d, String source) {

   237         if (cs == null) {

   238             d.dispose();

   239         } else {

   240             cs.whenComplete((whatever, error) -> {

   241                 if (error != null) {

   242                     String m = "CompletionStage returned by " + source + " completed exceptionally";

   243                     closeExceptionally(new RuntimeException(m, error));

   244                 }

   245                 d.dispose();

   246             });

   247         }

   248     }

   249 

   250     private void decrementDemand() {

   251         long newDemand = demand.decrementAndGet();

   252         assert newDemand >= 0 : newDemand;

   253     }

   254 

   255     private RawChannel.RawEvent createChannelEvent() {

   256         return new RawChannel.RawEvent() {

   257 

   258             @Override

   259             public int interestOps() {

   260                 return SelectionKey.OP_READ;

   261             }

   262 

   263             @Override

   264             public void handle() {

   265                 boolean wasNotReadable = readable.compareAndSet(false, true);

   266                 assert wasNotReadable;

   267                 handler.signal();

   268             }

   269 

   270             @Override

   271             public String toString() {

   272                 return "Read readiness event [" + channel + "]";

   273             }

   274         };

   275     }

   276 }