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 java.io.IOException;

    29 import java.nio.channels.SelectionKey;

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

    31 import java.util.function.Consumer;

    32 

    33 import static java.util.Objects.requireNonNull;

    34 

    35 /*

    36  * Sends messages one at a time, in an asynchronous and non-blocking fashion.

    37  *

    38  * No matter whether the message has been fully sent or an error has occurred,

    39  * the transmitter reports the outcome to the supplied handler and becomes ready

    40  * to accept a new message. Until then, the transmitter is considered "busy" and

    41  * an IllegalStateException will be thrown on each attempt to invoke send.

    42  */

    43 public class Transmitter {

    44 

    45     /* This flag is used solely for assertions */

    46     private final AtomicBoolean busy = new AtomicBoolean();

    47     private OutgoingMessage message;

    48     private Consumer<Exception> completionHandler;

    49     private final RawChannel channel;

    50     private final RawChannel.RawEvent event;

    51 

    52     public Transmitter(RawChannel channel) {

    53         this.channel = channel;

    54         this.event = createHandler();

    55     }

    56 

    57     /**

    58      * The supplied handler may be invoked in the calling thread.

    59      * A {@code StackOverflowError} may thus occur if there's a possibility

    60      * that this method is called again by the supplied handler.

    61      */

    62     public void send(OutgoingMessage message,

    63                      Consumer<Exception> completionHandler)

    64     {

    65         requireNonNull(message);

    66         requireNonNull(completionHandler);

    67         if (!busy.compareAndSet(false, true)) {

    68             throw new IllegalStateException();

    69         }

    70         send0(message, completionHandler);

    71     }

    72 

    73     public void close() throws IOException {

    74         channel.shutdownOutput();

    75     }

    76 

    77     private RawChannel.RawEvent createHandler() {

    78         return new RawChannel.RawEvent() {

    79 

    80             @Override

    81             public int interestOps() {

    82                 return SelectionKey.OP_WRITE;

    83             }

    84 

    85             @Override

    86             public void handle() {

    87                 // registerEvent(e) happens-before subsequent e.handle(), so

    88                 // we're fine reading the stored message and the completionHandler

    89                 send0(message, completionHandler);

    90             }

    91         };

    92     }

    93 

    94     private void send0(OutgoingMessage message, Consumer<Exception> handler) {

    95         boolean b = busy.get();

    96         assert b; // Please don't inline this, as busy.get() has memory

    97                   // visibility effects and we don't want the program behaviour

    98                   // to depend on whether the assertions are turned on

    99                   // or turned off

   100         try {

   101             boolean sent = message.sendTo(channel);

   102             if (sent) {

   103                 busy.set(false);

   104                 handler.accept(null);

   105             } else {

   106                 // The message has not been fully sent, the transmitter needs to

   107                 // remember the message until it can continue with sending it

   108                 this.message = message;

   109                 this.completionHandler = handler;

   110                 try {

   111                     channel.registerEvent(event);

   112                 } catch (IOException e) {

   113                     this.message = null;

   114                     this.completionHandler = null;

   115                     busy.set(false);

   116                     handler.accept(e);

   117                 }

   118             }

   119         } catch (IOException e) {

   120             busy.set(false);

   121             handler.accept(e);

   122         }

   123     }

   124 }