1 /*

     2  * Copyright (c) 2015, 2017, 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 jdk.incubator.http.internal.websocket;

    27 

    28 import jdk.incubator.http.internal.common.Demand;

    29 import jdk.incubator.http.internal.common.SequentialScheduler;

    30 

    31 import java.io.IOException;

    32 import java.nio.ByteBuffer;

    33 import java.nio.channels.SelectionKey;

    34 

    35 /*

    36  * Receives incoming data from the channel on demand and converts it into a

    37  * stream of WebSocket messages which are then delivered to the supplied message

    38  * consumer in a strict sequential order and non-recursively. In other words,

    39  *

    40  *     onText()

    41  *     onText()

    42  *     onBinary()

    43  *     ...

    44  *

    45  * instead of

    46  *

    47  *     onText()

    48  *       onText()

    49  *         onBinary()

    50  *     ...

    51  *

    52  * even if `request(long n)` is called from inside these invocations.

    53  */

    54 public class Receiver {

    55 

    56     private final MessageStreamConsumer messageConsumer;

    57     private final RawChannel channel;

    58     private final FrameConsumer frameConsumer;

    59     private final Frame.Reader reader = new Frame.Reader();

    60     private final RawChannel.RawEvent event = createHandler();

    61     protected final Demand demand = new Demand(); /* Exposed for testing purposes */

    62     private final SequentialScheduler pushScheduler;

    63 

    64     private ByteBuffer data;

    65     private volatile int state;

    66 

    67     private static final int UNREGISTERED = 0;

    68     private static final int AVAILABLE    = 1;

    69     private static final int WAITING      = 2;

    70 

    71     public Receiver(MessageStreamConsumer messageConsumer, RawChannel channel) {

    72         this.messageConsumer = messageConsumer;

    73         this.channel = channel;

    74         this.frameConsumer = new FrameConsumer(this.messageConsumer);

    75         this.data = channel.initialByteBuffer();

    76         // To ensure the initial non-final `data` will be visible

    77         // (happens-before) when `handler` invokes `pushContinuously`

    78         // the following assignment is done last:

    79         pushScheduler = createScheduler();

    80     }

    81 

    82     /* Exposed for testing purposes */

    83     protected SequentialScheduler createScheduler() {

    84         return new SequentialScheduler(new PushContinuouslyTask());

    85     }

    86 

    87     private RawChannel.RawEvent createHandler() {

    88         return new RawChannel.RawEvent() {

    89 

    90             @Override

    91             public int interestOps() {

    92                 return SelectionKey.OP_READ;

    93             }

    94 

    95             @Override

    96             public void handle() {

    97                 state = AVAILABLE;

    98                 pushScheduler.runOrSchedule();

    99             }

   100         };

   101     }

   102 

   103     public void request(long n) {

   104         if (demand.increase(n)) {

   105             pushScheduler.runOrSchedule();

   106         }

   107     }

   108 

   109     /*

   110      * Why is this method needed? Since Receiver operates through callbacks

   111      * this method allows to abstract out what constitutes as a message being

   112      * received (i.e. to decide outside this type when exactly one should

   113      * decrement the demand).

   114      */

   115     void acknowledge() {

   116         long x = demand.decreaseAndGet(1);

   117         if (x < 0) {

   118             throw new InternalError(String.valueOf(x));

   119         }

   120     }

   121 

   122     /*

   123      * Stops the machinery from reading and delivering messages permanently,

   124      * regardless of the current demand and data availability.

   125      */

   126     public void close() throws IOException {

   127         pushScheduler.stop();

   128         channel.shutdownInput();

   129     }

   130 

   131     private class PushContinuouslyTask

   132         extends SequentialScheduler.CompleteRestartableTask

   133     {

   134         @Override

   135         public void run() {

   136             while (!pushScheduler.isStopped()) {

   137                 if (data.hasRemaining()) {

   138                     if (!demand.isFulfilled()) {

   139                         try {

   140                             int oldPos = data.position();

   141                             reader.readFrame(data, frameConsumer);

   142                             int newPos = data.position();

   143                             assert oldPos != newPos : data; // reader always consumes bytes

   144                         } catch (Throwable e) {

   145                             pushScheduler.stop();

   146                             messageConsumer.onError(e);

   147                         }

   148                         continue;

   149                     }

   150                     break;

   151                 }

   152                 switch (state) {

   153                     case WAITING:

   154                         return;

   155                     case UNREGISTERED:

   156                         try {

   157                             state = WAITING;

   158                             channel.registerEvent(event);

   159                         } catch (Throwable e) {

   160                             pushScheduler.stop();

   161                             messageConsumer.onError(e);

   162                         }

   163                         return;

   164                     case AVAILABLE:

   165                         try {

   166                             data = channel.read();

   167                         } catch (Throwable e) {

   168                             pushScheduler.stop();

   169                             messageConsumer.onError(e);

   170                             return;

   171                         }

   172                         if (data == null) { // EOF

   173                             pushScheduler.stop();

   174                             messageConsumer.onComplete();

   175                             return;

   176                         } else if (!data.hasRemaining()) { // No data at the moment

   177                             // Pretty much a "goto", reusing the existing code path

   178                             // for registration

   179                             state = UNREGISTERED;

   180                         }

   181                         continue;

   182                     default:

   183                         throw new InternalError(String.valueOf(state));

   184                 }

   185             }

   186         }

   187     }

   188 }