1 /*

     2  * Copyright (c) 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.

     8  *

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

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

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

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

    13  * accompanied this code).

    14  *

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

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

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

    18  *

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

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

    21  * questions.

    22  */

    23 

    24 package jdk.incubator.http.internal.websocket;

    25 

    26 import jdk.incubator.http.WebSocket;

    27 import org.testng.annotations.Test;

    28 

    29 import java.net.URI;

    30 import java.nio.ByteBuffer;

    31 import java.util.Random;

    32 import java.util.concurrent.CompletableFuture;

    33 import java.util.concurrent.CompletionStage;

    34 import java.util.concurrent.TimeUnit;

    35 import java.util.concurrent.atomic.AtomicInteger;

    36 

    37 import static jdk.incubator.http.WebSocket.NORMAL_CLOSURE;

    38 import static jdk.incubator.http.internal.websocket.TestSupport.assertCompletesExceptionally;

    39 import static jdk.incubator.http.internal.websocket.WebSocketImpl.newInstance;

    40 import static org.testng.Assert.assertEquals;

    41 

    42 public class SendingTest {

    43 

    44     @Test

    45     public void sendTextImmediately() {

    46         MockTransmitter t = new MockTransmitter() {

    47             @Override

    48             protected CompletionStage<?> whenSent() {

    49                 return CompletableFuture.completedFuture(null);

    50             }

    51         };

    52         WebSocket ws = newWebSocket(t);

    53         CompletableFuture.completedFuture(ws)

    54                 .thenCompose(w -> w.sendText("1", true))

    55                 .thenCompose(w -> w.sendText("2", true))

    56                 .thenCompose(w -> w.sendText("3", true))

    57                 .join();

    58 

    59         assertEquals(t.queue().size(), 3);

    60     }

    61 

    62     @Test

    63     public void sendTextWithDelay() {

    64         MockTransmitter t = new MockTransmitter() {

    65             @Override

    66             protected CompletionStage<?> whenSent() {

    67                 return new CompletableFuture<>()

    68                         .completeOnTimeout(null, 1, TimeUnit.SECONDS);

    69             }

    70         };

    71         WebSocket ws = newWebSocket(t);

    72         CompletableFuture.completedFuture(ws)

    73                 .thenCompose(w -> w.sendText("1", true))

    74                 .thenCompose(w -> w.sendText("2", true))

    75                 .thenCompose(w -> w.sendText("3", true))

    76                 .join();

    77 

    78         assertEquals(t.queue().size(), 3);

    79     }

    80 

    81     @Test

    82     public void sendTextMixedDelay() {

    83         Random r = new Random();

    84 

    85         MockTransmitter t = new MockTransmitter() {

    86             @Override

    87             protected CompletionStage<?> whenSent() {

    88                 return r.nextBoolean() ?

    89                         new CompletableFuture<>().completeOnTimeout(null, 1, TimeUnit.SECONDS) :

    90                         CompletableFuture.completedFuture(null);

    91             }

    92         };

    93         WebSocket ws = newWebSocket(t);

    94         CompletableFuture.completedFuture(ws)

    95                 .thenCompose(w -> w.sendText("1", true))

    96                 .thenCompose(w -> w.sendText("2", true))

    97                 .thenCompose(w -> w.sendText("3", true))

    98                 .thenCompose(w -> w.sendText("4", true))

    99                 .thenCompose(w -> w.sendText("5", true))

   100                 .thenCompose(w -> w.sendText("6", true))

   101                 .thenCompose(w -> w.sendText("7", true))

   102                 .thenCompose(w -> w.sendText("8", true))

   103                 .thenCompose(w -> w.sendText("9", true))

   104                 .join();

   105 

   106         assertEquals(t.queue().size(), 9);

   107     }

   108 

   109     @Test

   110     public void sendControlMessagesConcurrently() {

   111 

   112         CompletableFuture<?> first = new CompletableFuture<>();

   113 

   114         MockTransmitter t = new MockTransmitter() {

   115 

   116             final AtomicInteger i = new AtomicInteger();

   117 

   118             @Override

   119             protected CompletionStage<?> whenSent() {

   120                 if (i.incrementAndGet() == 1) {

   121                     return first;

   122                 } else {

   123                     return CompletableFuture.completedFuture(null);

   124                 }

   125             }

   126         };

   127         WebSocket ws = newWebSocket(t);

   128 

   129         CompletableFuture<?> cf1 = ws.sendPing(ByteBuffer.allocate(0));

   130         CompletableFuture<?> cf2 = ws.sendPong(ByteBuffer.allocate(0));

   131         CompletableFuture<?> cf3 = ws.sendClose(NORMAL_CLOSURE, "");

   132         CompletableFuture<?> cf4 = ws.sendClose(NORMAL_CLOSURE, "");

   133         CompletableFuture<?> cf5 = ws.sendPing(ByteBuffer.allocate(0));

   134         CompletableFuture<?> cf6 = ws.sendPong(ByteBuffer.allocate(0));

   135 

   136         first.complete(null);

   137         // Don't care about exceptional completion, only that all of them have

   138         // completed

   139         CompletableFuture.allOf(cf1, cf2, cf3, cf4, cf5, cf6)

   140                 .handle((v, e) -> null).join();

   141 

   142         cf3.join(); /* Check that sendClose has completed normally */

   143         cf4.join(); /* Check that repeated sendClose has completed normally */

   144         assertCompletesExceptionally(IllegalStateException.class, cf5);

   145         assertCompletesExceptionally(IllegalStateException.class, cf6);

   146 

   147         assertEquals(t.queue().size(), 3); // 6 minus 3 that were not accepted

   148     }

   149 

   150     private static WebSocket newWebSocket(Transmitter transmitter) {

   151         URI uri = URI.create("ws://localhost");

   152         String subprotocol = "";

   153         TransportSupplier transport = new MockTransport() {

   154             @Override

   155             public Transmitter transmitter() {

   156                 return transmitter;

   157             }

   158         };

   159         return newInstance(uri,

   160                            subprotocol,

   161                            new MockListener(Long.MAX_VALUE),

   162                            transport);

   163     }

   164 }