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.internal.net.http.websocket;

    25 

    26 import java.net.http.WebSocket.MessagePart;

    27 import jdk.internal.net.http.common.Demand;

    28 import jdk.internal.net.http.common.SequentialScheduler;

    29 

    30 import java.io.IOException;

    31 import java.nio.ByteBuffer;

    32 import java.util.Collection;

    33 import java.util.LinkedList;

    34 import java.util.List;

    35 import java.util.Objects;

    36 import java.util.Queue;

    37 import java.util.concurrent.CompletableFuture;

    38 import java.util.concurrent.ConcurrentLinkedQueue;

    39 import java.util.function.Consumer;

    40 import java.util.function.Supplier;

    41 

    42 import static jdk.internal.net.http.websocket.TestSupport.fullCopy;

    43 

    44 public class MockTransport<T> implements Transport<T> {

    45 

    46     private final long startTime = System.currentTimeMillis();

    47     private final Queue<Invocation> output = new ConcurrentLinkedQueue<>();

    48     private final Queue<CompletableFuture<Consumer<MessageStreamConsumer>>>

    49             input = new ConcurrentLinkedQueue<>();

    50     private final Supplier<T> supplier;

    51     private final MessageStreamConsumer consumer;

    52     private final SequentialScheduler scheduler

    53             = new SequentialScheduler(new ReceiveTask());

    54     private final Demand demand = new Demand();

    55 

    56     public MockTransport(Supplier<T> sendResultSupplier,

    57                          MessageStreamConsumer consumer) {

    58         this.supplier = sendResultSupplier;

    59         this.consumer = consumer;

    60         input.addAll(receive());

    61     }

    62 

    63     @Override

    64     public final CompletableFuture<T> sendText(CharSequence message,

    65                                                boolean isLast) {

    66         output.add(Invocation.sendText(message, isLast));

    67         return send(String.format("sendText(%s, %s)", message, isLast),

    68                     () -> sendText0(message, isLast));

    69     }

    70 

    71     protected CompletableFuture<T> sendText0(CharSequence message,

    72                                              boolean isLast) {

    73         return defaultSend();

    74     }

    75 

    76     protected CompletableFuture<T> defaultSend() {

    77         return CompletableFuture.completedFuture(result());

    78     }

    79 

    80     @Override

    81     public final CompletableFuture<T> sendBinary(ByteBuffer message,

    82                                                  boolean isLast) {

    83         output.add(Invocation.sendBinary(message, isLast));

    84         return send(String.format("sendBinary(%s, %s)", message, isLast),

    85                     () -> sendBinary0(message, isLast));

    86     }

    87 

    88     protected CompletableFuture<T> sendBinary0(ByteBuffer message,

    89                                                boolean isLast) {

    90         return defaultSend();

    91     }

    92 

    93     @Override

    94     public final CompletableFuture<T> sendPing(ByteBuffer message) {

    95         output.add(Invocation.sendPing(message));

    96         return send(String.format("sendPing(%s)", message),

    97                     () -> sendPing0(message));

    98     }

    99 

   100     protected CompletableFuture<T> sendPing0(ByteBuffer message) {

   101         return defaultSend();

   102     }

   103 

   104     @Override

   105     public final CompletableFuture<T> sendPong(ByteBuffer message) {

   106         output.add(Invocation.sendPong(message));

   107         return send(String.format("sendPong(%s)", message),

   108                     () -> sendPong0(message));

   109     }

   110 

   111     protected CompletableFuture<T> sendPong0(ByteBuffer message) {

   112         return defaultSend();

   113     }

   114 

   115     @Override

   116     public final CompletableFuture<T> sendClose(int statusCode, String reason) {

   117         output.add(Invocation.sendClose(statusCode, reason));

   118         return send(String.format("sendClose(%s, %s)", statusCode, reason),

   119                     () -> sendClose0(statusCode, reason));

   120     }

   121 

   122     protected CompletableFuture<T> sendClose0(int statusCode, String reason) {

   123         return defaultSend();

   124     }

   125 

   126     protected Collection<CompletableFuture<Consumer<MessageStreamConsumer>>> receive() {

   127         return List.of();

   128     }

   129 

   130     public static Consumer<MessageStreamConsumer> onText(CharSequence data,

   131                                                          MessagePart part) {

   132         return c -> c.onText(data.toString(), part);

   133     }

   134 

   135     public static Consumer<MessageStreamConsumer> onBinary(ByteBuffer data,

   136                                                            MessagePart part) {

   137         return c -> c.onBinary(fullCopy(data), part);

   138     }

   139 

   140     public static Consumer<MessageStreamConsumer> onPing(ByteBuffer data) {

   141         return c -> c.onPing(fullCopy(data));

   142     }

   143 

   144     public static Consumer<MessageStreamConsumer> onPong(ByteBuffer data) {

   145         return c -> c.onPong(fullCopy(data));

   146     }

   147 

   148     public static Consumer<MessageStreamConsumer> onClose(int statusCode,

   149                                                           String reason) {

   150         return c -> c.onClose(statusCode, reason);

   151     }

   152 

   153     public static Consumer<MessageStreamConsumer> onError(Throwable error) {

   154         return c -> c.onError(error);

   155     }

   156 

   157     public static Consumer<MessageStreamConsumer> onComplete() {

   158         return c -> c.onComplete();

   159     }

   160 

   161     @Override

   162     public void request(long n) {

   163         demand.increase(n);

   164         scheduler.runOrSchedule();

   165     }

   166 

   167     @Override

   168     public void acknowledgeReception() {

   169         demand.tryDecrement();

   170     }

   171 

   172     @Override

   173     public final void closeOutput() throws IOException {

   174         output.add(Invocation.closeOutput());

   175         begin("closeOutput()");

   176         closeOutput0();

   177         end("closeOutput()");

   178     }

   179 

   180     protected void closeOutput0() throws IOException {

   181         defaultClose();

   182     }

   183 

   184     protected void defaultClose() throws IOException {

   185     }

   186 

   187     @Override

   188     public final void closeInput() throws IOException {

   189         output.add(Invocation.closeInput());

   190         begin("closeInput()");

   191         closeInput0();

   192         end("closeInput()");

   193     }

   194 

   195     protected void closeInput0() throws IOException {

   196         defaultClose();

   197     }

   198 

   199     public abstract static class Invocation {

   200 

   201         static Invocation.SendText sendText(CharSequence message,

   202                                             boolean isLast) {

   203             return new SendText(message, isLast);

   204         }

   205 

   206         static Invocation.SendBinary sendBinary(ByteBuffer message,

   207                                                 boolean isLast) {

   208             return new SendBinary(message, isLast);

   209         }

   210 

   211         static Invocation.SendPing sendPing(ByteBuffer message) {

   212             return new SendPing(message);

   213         }

   214 

   215         static Invocation.SendPong sendPong(ByteBuffer message) {

   216             return new SendPong(message);

   217         }

   218 

   219         static Invocation.SendClose sendClose(int statusCode, String reason) {

   220             return new SendClose(statusCode, reason);

   221         }

   222 

   223         public static CloseOutput closeOutput() {

   224             return new CloseOutput();

   225         }

   226 

   227         public static CloseInput closeInput() {

   228             return new CloseInput();

   229         }

   230 

   231         public static final class SendText extends Invocation {

   232 

   233             final CharSequence message;

   234             final boolean isLast;

   235 

   236             SendText(CharSequence message, boolean isLast) {

   237                 this.message = message.toString();

   238                 this.isLast = isLast;

   239             }

   240 

   241             @Override

   242             public boolean equals(Object obj) {

   243                 if (this == obj) return true;

   244                 if (obj == null || getClass() != obj.getClass()) return false;

   245                 SendText sendText = (SendText) obj;

   246                 return isLast == sendText.isLast &&

   247                         Objects.equals(message, sendText.message);

   248             }

   249 

   250             @Override

   251             public int hashCode() {

   252                 return Objects.hash(isLast, message);

   253             }

   254         }

   255 

   256         public static final class SendBinary extends Invocation {

   257 

   258             final ByteBuffer message;

   259             final boolean isLast;

   260 

   261             SendBinary(ByteBuffer message, boolean isLast) {

   262                 this.message = fullCopy(message);

   263                 this.isLast = isLast;

   264             }

   265 

   266             @Override

   267             public boolean equals(Object obj) {

   268                 if (this == obj) return true;

   269                 if (obj == null || getClass() != obj.getClass()) return false;

   270                 SendBinary that = (SendBinary) obj;

   271                 return isLast == that.isLast &&

   272                         Objects.equals(message, that.message);

   273             }

   274 

   275             @Override

   276             public int hashCode() {

   277                 return Objects.hash(message, isLast);

   278             }

   279         }

   280 

   281         private static final class SendPing extends Invocation {

   282 

   283             final ByteBuffer message;

   284 

   285             SendPing(ByteBuffer message) {

   286                 this.message = fullCopy(message);

   287             }

   288 

   289             @Override

   290             public boolean equals(Object obj) {

   291                 if (this == obj) return true;

   292                 if (obj == null || getClass() != obj.getClass()) return false;

   293                 SendPing sendPing = (SendPing) obj;

   294                 return Objects.equals(message, sendPing.message);

   295             }

   296 

   297             @Override

   298             public int hashCode() {

   299                 return Objects.hash(message);

   300             }

   301         }

   302 

   303         private static final class SendPong extends Invocation {

   304 

   305             final ByteBuffer message;

   306 

   307             SendPong(ByteBuffer message) {

   308                 this.message = fullCopy(message);

   309             }

   310 

   311             @Override

   312             public boolean equals(Object obj) {

   313                 if (this == obj) return true;

   314                 if (obj == null || getClass() != obj.getClass()) return false;

   315                 SendPing sendPing = (SendPing) obj;

   316                 return Objects.equals(message, sendPing.message);

   317             }

   318 

   319             @Override

   320             public int hashCode() {

   321                 return Objects.hash(message);

   322             }

   323         }

   324 

   325         private static final class SendClose extends Invocation {

   326 

   327             final int statusCode;

   328             final String reason;

   329 

   330             SendClose(int statusCode, String reason) {

   331                 this.statusCode = statusCode;

   332                 this.reason = reason;

   333             }

   334 

   335             @Override

   336             public boolean equals(Object obj) {

   337                 if (this == obj) return true;

   338                 if (obj == null || getClass() != obj.getClass()) return false;

   339                 SendClose sendClose = (SendClose) obj;

   340                 return statusCode == sendClose.statusCode &&

   341                         Objects.equals(reason, sendClose.reason);

   342             }

   343 

   344             @Override

   345             public int hashCode() {

   346                 return Objects.hash(statusCode, reason);

   347             }

   348         }

   349 

   350         private static final class CloseOutput extends Invocation {

   351 

   352             CloseOutput() { }

   353 

   354             @Override

   355             public int hashCode() {

   356                 return 0;

   357             }

   358 

   359             @Override

   360             public boolean equals(Object obj) {

   361                 return obj instanceof CloseOutput;

   362             }

   363         }

   364 

   365         private static final class CloseInput extends Invocation {

   366 

   367             CloseInput() { }

   368 

   369             @Override

   370             public int hashCode() {

   371                 return 0;

   372             }

   373 

   374             @Override

   375             public boolean equals(Object obj) {

   376                 return obj instanceof CloseInput;

   377             }

   378         }

   379     }

   380 

   381     public Queue<Invocation> invocations() {

   382         return new LinkedList<>(output);

   383     }

   384 

   385     protected final T result() {

   386         return supplier.get();

   387     }

   388 

   389     private CompletableFuture<T> send(String name,

   390                                       Supplier<CompletableFuture<T>> supplier) {

   391         begin(name);

   392         CompletableFuture<T> cf = supplier.get().whenComplete((r, e) -> {

   393             System.out.printf("[%6s ms.] complete %s%n", elapsedTime(), name);

   394         });

   395         end(name);

   396         return cf;

   397     }

   398 

   399     private void begin(String name) {

   400         System.out.printf("[%6s ms.] begin %s%n", elapsedTime(), name);

   401     }

   402 

   403     private void end(String name) {

   404         System.out.printf("[%6s ms.] end %s%n", elapsedTime(), name);

   405     }

   406 

   407     private long elapsedTime() {

   408         return System.currentTimeMillis() - startTime;

   409     }

   410 

   411     private final class ReceiveTask implements SequentialScheduler.RestartableTask {

   412 

   413         @Override

   414         public void run(SequentialScheduler.DeferredCompleter taskCompleter) {

   415             if (!scheduler.isStopped() && !demand.isFulfilled() && !input.isEmpty()) {

   416                 CompletableFuture<Consumer<MessageStreamConsumer>> cf = input.remove();

   417                 if (cf.isDone()) { // Forcing synchronous execution

   418                     cf.join().accept(consumer);

   419                     repeat(taskCompleter);

   420                 } else {

   421                     cf.whenCompleteAsync((r, e) -> {

   422                         r.accept(consumer);

   423                         repeat(taskCompleter);

   424                     });

   425                 }

   426             } else {

   427                 taskCompleter.complete();

   428             }

   429         }

   430 

   431         private void repeat(SequentialScheduler.DeferredCompleter taskCompleter) {

   432             taskCompleter.complete();

   433             scheduler.runOrSchedule();

   434         }

   435     }

   436 }