jdk-sandbox: comparison test/jdk/java/net/httpclient/whitebox/jdk.incubator.httpclient/jdk/incubator/http/SSLTubeTest.java

equal deleted inserted replaced

-:2d423c9b73bb
+:8d4770c22b63
 import java.security.NoSuchAlgorithmException;
 import java.security.UnrecoverableKeyException;
 import java.security.cert.CertificateException;
 import java.util.List;
 import java.util.Queue;
+import java.util.Random;
 import java.util.StringTokenizer;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.SubmissionPublisher;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 @Test
 public class SSLTubeTest {
 private static final long COUNTER = 600;
 private static final int LONGS_PER_BUF = 800;
 private static final long TOTAL_LONGS = COUNTER * LONGS_PER_BUF;
+public static final ByteBuffer SENTINEL = ByteBuffer.allocate(0);
+static final Random rand = new Random();
+static int randomRange(int lower, int upper) {
+if (lower > upper)
+throw new IllegalArgumentException("lower > upper");
+int diff = upper - lower;
+int r = lower + rand.nextInt(diff);
+return r - (r % 8); // round down to multiple of 8 (align for longs)
+}
 private static ByteBuffer getBuffer(long startingAt) {
 ByteBuffer buf = ByteBuffer.allocate(LONGS_PER_BUF * 8);
 for (int j = 0; j < LONGS_PER_BUF; j++) {
 buf.putLong(startingAt++);
 }
 buf.flip();
 return buf;
 }
-@Test(timeOut = 30000)
+@Test
-public void run() throws IOException {
+public void runWithSSLLoopackServer() throws IOException {
+ExecutorService sslExecutor = Executors.newCachedThreadPool();
 /* Start of wiring */
-ExecutorService sslExecutor = Executors.newCachedThreadPool();
 /* Emulates an echo server */
-//        FlowTube server = new SSLTube(createSSLEngine(false),
-//                                      sslExecutor,
-//                                      new EchoTube(16));
 SSLLoopbackSubscriber server =
 new SSLLoopbackSubscriber((new SimpleSSLContext()).get(), sslExecutor);
 server.start();
+run(server, sslExecutor);
+}
+@Test
+public void runWithEchoServer() throws IOException {
+ExecutorService sslExecutor = Executors.newCachedThreadPool();
+/* Start of wiring */
+/* Emulates an echo server */
+FlowTube server = crossOverEchoServer(sslExecutor);
+run(server, sslExecutor);
+}
+private void run(FlowTube server, ExecutorService sslExecutor) throws IOException {
 FlowTube client = new SSLTube(createSSLEngine(true),
 sslExecutor,
 server);
 SubmissionPublisher<List<ByteBuffer>> p =
 new SubmissionPublisher<>(ForkJoinPool.commonPool(),
 } finally {
 sslExecutor.shutdownNow();
 }
 }
+/**
+* This is a copy of the SSLLoopbackSubscriber used in FlowTest
+*/
 static class SSLLoopbackSubscriber implements FlowTube {
 private final BlockingQueue<ByteBuffer> buffer;
 private final Socket clientSock;
 private final SSLSocket serverSock;
 private final Thread thread1, thread2, thread3;
 // reads off the SSLSocket the data from the "server"
 private void clientReader() {
 try {
 InputStream is = clientSock.getInputStream();
-final int bufsize = FlowTest.randomRange(512, 16 * 1024);
+final int bufsize = randomRange(512, 16 * 1024);
 System.out.println("clientReader: bufsize = " + bufsize);
 while (true) {
 byte[] buf = new byte[bufsize];
 int n = is.read(buf);
 if (n == -1) {
 OutputStream os =
 new BufferedOutputStream(clientSock.getOutputStream());
 while (true) {
 ByteBuffer buf = buffer.take();
-if (buf == FlowTest.SENTINEL) {
+if (buf == SENTINEL) {
 // finished
 //Utils.sleep(2000);
 System.out.println("clientWriter close: " + nbytes + " written");
 clientSock.shutdownOutput();
 System.out.println("clientWriter close return");
 // thread2
 private void serverLoopback() {
 try {
 InputStream is = serverSock.getInputStream();
 OutputStream os = serverSock.getOutputStream();
-final int bufsize = FlowTest.randomRange(512, 16 * 1024);
+final int bufsize = randomRange(512, 16 * 1024);
 System.out.println("serverLoopback: bufsize = " + bufsize);
 byte[] bb = new byte[bufsize];
 while (true) {
 int n = is.read(bb);
 if (n == -1) {
 }
 @Override
 public void onComplete() {
 try {
-buffer.put(FlowTest.SENTINEL);
+buffer.put(SENTINEL);
 } catch (InterruptedException e) {
 e.printStackTrace();
 Utils.close(clientSock);
 }
 }
 Thread.sleep(millis);
 } catch (InterruptedException e) {
 }
 }
-//    private static final class EchoTube implements FlowTube {
-//
+/**
-//        private final static Object EOF = new Object();
+* Creates a cross-over FlowTube than can be plugged into a client-side
-//        private final Executor executor = Executors.newSingleThreadExecutor();
+* SSLTube (in place of the SSLLoopbackSubscriber).
-//
+* Note that the only method that can be called on the return tube
-//        private final Queue<Object> queue = new ConcurrentLinkedQueue<>();
+* is connectFlows(). Calling any other method will trigger an
-//        private final int maxQueueSize;
+* InternalError.
-//        private final SequentialScheduler processingScheduler =
+* @param sslExecutor an executor
-//                new SequentialScheduler(createProcessingTask());
+* @return a cross-over FlowTube connected to an EchoTube.
-//
+* @throws IOException
-//        /* Writing into this tube */
+*/
-//        private long unfulfilled;
+FlowTube crossOverEchoServer(Executor sslExecutor) throws IOException {
-//        private Flow.Subscription subscription;
+LateBindingTube crossOver = new LateBindingTube();
-//
+FlowTube server = new SSLTube(createSSLEngine(false),
-//        /* Reading from this tube */
+sslExecutor,
-//        private final Demand demand = new Demand();
+crossOver);
-//        private final AtomicBoolean cancelled = new AtomicBoolean();
+EchoTube echo = new EchoTube(6);
-//        private Flow.Subscriber<? super List<ByteBuffer>> subscriber;
+server.connectFlows(FlowTube.asTubePublisher(echo), FlowTube.asTubeSubscriber(echo));
-//
-//        private EchoTube(int maxBufferSize) {
+return new CrossOverTube(crossOver);
-//            if (maxBufferSize < 1)
+}
-//                throw new IllegalArgumentException();
-//            this.maxQueueSize = maxBufferSize;
+/**
-//        }
+* A cross-over FlowTube that makes it possible to reverse the direction
-//
+* of flows. The typical usage is to connect an two opposite SSLTube,
-//        @Override
+* one encrypting, one decrypting, to e.g. an EchoTube, with the help
-//        public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
+* of a LateBindingTube:
-//            this.subscriber = subscriber;
+* {@code
-//            System.out.println("EchoTube got subscriber: " + subscriber);
+* client app => SSLTube => CrossOverTube <= LateBindingTube <= SSLTube <= EchoTube
-//            this.subscriber.onSubscribe(new InternalSubscription());
+* }
-//        }
+* <p>
-//
+* Note that the only method that can be called on the CrossOverTube is
-//        @Override
+* connectFlows(). Calling any other method will cause an InternalError to
-//        public void onSubscribe(Flow.Subscription subscription) {
+* be thrown.
-//            unfulfilled = maxQueueSize;
+* Also connectFlows() can be called only once.
-//            System.out.println("EchoTube request: " + maxQueueSize);
+*/
-//            (this.subscription = subscription).request(maxQueueSize);
+private static final class CrossOverTube implements FlowTube {
-//        }
+final LateBindingTube tube;
-//
+CrossOverTube(LateBindingTube tube) {
-//        @Override
+this.tube = tube;
-//        public void onNext(List<ByteBuffer> item) {
+}
-//            if (--unfulfilled == (maxQueueSize / 2)) {
-//                long req = maxQueueSize - unfulfilled;
+@Override
-//                subscription.request(req);
+public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
-//                System.out.println("EchoTube request: " + req);
+throw newInternalError();
-//                unfulfilled = maxQueueSize;
+}
-//            }
-//            System.out.println("EchoTube add " + Utils.remaining(item));
+@Override
-//            queue.add(item);
+public void connectFlows(TubePublisher writePublisher, TubeSubscriber readSubscriber) {
-//            processingScheduler.deferOrSchedule(executor);
+tube.start(writePublisher, readSubscriber);
-//        }
+}
-//
-//        @Override
+@Override
-//        public void onError(Throwable throwable) {
+public boolean isFinished() {
-//            System.out.println("EchoTube add " + throwable);
+return tube.isFinished();
-//            queue.add(throwable);
+}
-//            processingScheduler.deferOrSchedule(executor);
-//        }
+Error newInternalError() {
-//
+InternalError error = new InternalError();
-//        @Override
+error.printStackTrace(System.out);
-//        public void onComplete() {
+return error;
-//            System.out.println("EchoTube add EOF");
+}
-//            queue.add(EOF);
-//            processingScheduler.deferOrSchedule(executor);
+@Override
-//        }
+public void onSubscribe(Flow.Subscription subscription) {
-//
+throw newInternalError();
-//        @Override
+}
-//        public boolean isFinished() {
-//            return false;
+@Override
-//        }
+public void onError(Throwable throwable) {
-//
+throw newInternalError();
-//        private class InternalSubscription implements Flow.Subscription {
+}
-//
-//            @Override
+@Override
-//            public void request(long n) {
+public void onComplete() {
-//                System.out.println("EchoTube got request: " + n);
+throw newInternalError();
-//                if (n <= 0) {
+}
-//                    throw new InternalError();
-//                }
+@Override
-//                demand.increase(n);
+public void onNext(List<ByteBuffer> item) {
-//                processingScheduler.runOrSchedule();
+throw newInternalError();
-//            }
+}
-//
+}
-//            @Override
-//            public void cancel() {
+/**
-//                cancelled.set(true);
+* A late binding tube that makes it possible to create an
-//            }
+* SSLTube before the right-hand-side tube has been created.
-//        }
+* The typical usage is to make it possible to connect two
-//
+* opposite SSLTube (one encrypting, one decrypting) through a
-//        @Override
+* CrossOverTube:
-//        public String toString() {
+* {@code
-//            return "EchoTube";
+* client app => SSLTube => CrossOverTube <= LateBindingTube <= SSLTube <= EchoTube
-//        }
+* }
-//
+* <p>
-//        private SequentialScheduler.RestartableTask createProcessingTask() {
+* Note that this class only supports a single call to start(): it cannot be
-//            return new SequentialScheduler.CompleteRestartableTask() {
+* subscribed more than once from its left-hand-side (the cross over tube side).
-//
+*/
-//                @Override
+private static class LateBindingTube implements FlowTube {
-//                protected void run() {
-//                    try {
+final CompletableFuture<Flow.Publisher<List<ByteBuffer>>> futurePublisher
-//                        while (!cancelled.get()) {
+= new CompletableFuture<>();
-//                            Object item = queue.peek();
+final ConcurrentLinkedQueue<Consumer<Flow.Subscriber<? super List<ByteBuffer>>>> queue
-//                            if (item == null)
+= new ConcurrentLinkedQueue<>();
-//                                return;
+AtomicReference<Flow.Subscriber<? super List<ByteBuffer>>> subscriberRef = new AtomicReference<>();
-//                            try {
+SequentialScheduler scheduler = SequentialScheduler.synchronizedScheduler(this::loop);
-//                                System.out.println("EchoTube processing item");
+AtomicReference<Throwable> errorRef = new AtomicReference<>();
-//                                if (item instanceof List) {
+private volatile boolean finished;
-//                                    if (!demand.tryDecrement()) {
+private volatile boolean completed;
-//                                        System.out.println("EchoTube no demand");
-//                                        return;
-//                                    }
+public void start(Flow.Publisher<List<ByteBuffer>> publisher,
-//                                    @SuppressWarnings("unchecked")
+Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
-//                                    List<ByteBuffer> bytes = (List<ByteBuffer>) item;
+subscriberRef.set(subscriber);
-//                                    Object removed = queue.remove();
+futurePublisher.complete(publisher);
-//                                    assert removed == item;
+scheduler.runOrSchedule();
-//                                    System.out.println("EchoTube processing "
+}
-//                                            + Utils.remaining(bytes));
-//                                    subscriber.onNext(bytes);
+@Override
-//                                } else if (item instanceof Throwable) {
+public boolean isFinished() {
-//                                    cancelled.set(true);
+return finished;
-//                                    Object removed = queue.remove();
+}
-//                                    assert removed == item;
-//                                    System.out.println("EchoTube processing " + item);
+@Override
-//                                    subscriber.onError((Throwable) item);
+public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
-//                                } else if (item == EOF) {
+futurePublisher.thenAccept((p) -> p.subscribe(subscriber));
-//                                    cancelled.set(true);
+scheduler.runOrSchedule();
-//                                    Object removed = queue.remove();
+}
-//                                    assert removed == item;
-//                                    System.out.println("EchoTube processing EOF");
+@Override
-//                                    subscriber.onComplete();
+public void onSubscribe(Flow.Subscription subscription) {
-//                                } else {
+queue.add((s) -> s.onSubscribe(subscription));
-//                                    throw new InternalError(String.valueOf(item));
+scheduler.runOrSchedule();
-//                                }
+}
-//                            } finally {
-//                            }
+@Override
-//                        }
+public void onNext(List<ByteBuffer> item) {
-//                    } catch(Throwable t) {
+queue.add((s) -> s.onNext(item));
-//                        t.printStackTrace();
+scheduler.runOrSchedule();
-//                        throw t;
+}
-//                    }
-//                }
+@Override
-//            };
+public void onError(Throwable throwable) {
-//        }
+System.out.println("LateBindingTube onError");
-//    }
+throwable.printStackTrace(System.out);
+queue.add((s) -> {
+errorRef.compareAndSet(null, throwable);
+try {
+System.out.println("LateBindingTube subscriber onError: " + throwable);
+s.onError(errorRef.get());
+} finally {
+finished = true;
+System.out.println("LateBindingTube finished");
+}
+});
+scheduler.runOrSchedule();
+}
+@Override
+public void onComplete() {
+System.out.println("LateBindingTube completing");
+queue.add((s) -> {
+completed = true;
+try {
+System.out.println("LateBindingTube complete subscriber");
+s.onComplete();
+} finally {
+finished = true;
+System.out.println("LateBindingTube finished");
+}
+});
+scheduler.runOrSchedule();
+}
+private void loop() {
+if (finished) {
+scheduler.stop();
+return;
+}
+Flow.Subscriber<? super List<ByteBuffer>> subscriber = subscriberRef.get();
+if (subscriber == null) return;
+try {
+Consumer<Flow.Subscriber<? super List<ByteBuffer>>> s;
+while ((s = queue.poll()) != null) {
+s.accept(subscriber);
+}
+} catch (Throwable t) {
+if (errorRef.compareAndSet(null, t)) {
+onError(t);
+}
+}
+}
+}
+/**
+* An echo tube that just echoes back whatever bytes it receives.
+* This cannot be plugged to the right-hand-side of an SSLTube
+* since handshake data cannot be simply echoed back, and
+* application data most likely also need to be decrypted and
+* re-encrypted.
+*/
+private static final class EchoTube implements FlowTube {
+private final static Object EOF = new Object();
+private final Executor executor = Executors.newSingleThreadExecutor();
+private final Queue<Object> queue = new ConcurrentLinkedQueue<>();
+private final int maxQueueSize;
+private final SequentialScheduler processingScheduler =
+new SequentialScheduler(createProcessingTask());
+/* Writing into this tube */
+private volatile long requested;
+private Flow.Subscription subscription;
+/* Reading from this tube */
+private final Demand demand = new Demand();
+private final AtomicBoolean cancelled = new AtomicBoolean();
+private Flow.Subscriber<? super List<ByteBuffer>> subscriber;
+private EchoTube(int maxBufferSize) {
+if (maxBufferSize < 1)
+throw new IllegalArgumentException();
+this.maxQueueSize = maxBufferSize;
+}
+@Override
+public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {
+this.subscriber = subscriber;
+System.out.println("EchoTube got subscriber: " + subscriber);
+this.subscriber.onSubscribe(new InternalSubscription());
+}
+@Override
+public void onSubscribe(Flow.Subscription subscription) {
+System.out.println("EchoTube request: " + maxQueueSize);
+(this.subscription = subscription).request(requested = maxQueueSize);
+}
+private void requestMore() {
+Flow.Subscription s = subscription;
+if (s == null || cancelled.get()) return;
+long unfulfilled = queue.size() + --requested;
+if (unfulfilled <= maxQueueSize/2) {
+long req = maxQueueSize - unfulfilled;
+requested += req;
+s.request(req);
+System.out.printf("EchoTube request: %s [requested:%s, queue:%s, unfulfilled:%s]%n",
+req, requested-req, queue.size(), unfulfilled );
+}
+}
+@Override
+public void onNext(List<ByteBuffer> item) {
+System.out.printf("EchoTube add %s [requested:%s, queue:%s]%n",
+Utils.remaining(item), requested, queue.size());
+queue.add(item);
+processingScheduler.deferOrSchedule(executor);
+}
+@Override
+public void onError(Throwable throwable) {
+System.out.println("EchoTube add " + throwable);
+queue.add(throwable);
+processingScheduler.deferOrSchedule(executor);
+}
+@Override
+public void onComplete() {
+System.out.println("EchoTube add EOF");
+queue.add(EOF);
+processingScheduler.deferOrSchedule(executor);
+}
+@Override
+public boolean isFinished() {
+return cancelled.get();
+}
+private class InternalSubscription implements Flow.Subscription {
+@Override
+public void request(long n) {
+System.out.println("EchoTube got request: " + n);
+if (n <= 0) {
+throw new InternalError();
+}
+if (demand.increase(n)) {
+processingScheduler.deferOrSchedule(executor);
+}
+}
+@Override
+public void cancel() {
+cancelled.set(true);
+}
+}
+@Override
+public String toString() {
+return "EchoTube";
+}
+int transmitted = 0;
+private SequentialScheduler.RestartableTask createProcessingTask() {
+return new SequentialScheduler.CompleteRestartableTask() {
+@Override
+protected void run() {
+try {
+while (!cancelled.get()) {
+Object item = queue.peek();
+if (item == null) {
+System.out.printf("EchoTube: queue empty, requested=%s, demand=%s, transmitted=%s%n",
+requested, demand.get(), transmitted);
+requestMore();
+return;
+}
+try {
+System.out.printf("EchoTube processing item, requested=%s, demand=%s, transmitted=%s%n",
+requested, demand.get(), transmitted);
+if (item instanceof List) {
+if (!demand.tryDecrement()) {
+System.out.println("EchoTube no demand");
+return;
+}
+@SuppressWarnings("unchecked")
+List<ByteBuffer> bytes = (List<ByteBuffer>) item;
+Object removed = queue.remove();
+assert removed == item;
+System.out.println("EchoTube processing "
++ Utils.remaining(bytes));
+transmitted++;
+subscriber.onNext(bytes);
+requestMore();
+} else if (item instanceof Throwable) {
+cancelled.set(true);
+Object removed = queue.remove();
+assert removed == item;
+System.out.println("EchoTube processing " + item);
+subscriber.onError((Throwable) item);
+} else if (item == EOF) {
+cancelled.set(true);
+Object removed = queue.remove();
+assert removed == item;
+System.out.println("EchoTube processing EOF");
+subscriber.onComplete();
+} else {
+throw new InternalError(String.valueOf(item));
+}
+} finally {
+}
+}
+} catch(Throwable t) {
+t.printStackTrace();
+throw t;
+}
+}
+};
+}
+}
 /**
 * The final subscriber which receives the decrypted looped-back data. Just
 * needs to compare the data with what was sent. The given CF is either
 * completed exceptionally with an error or normally on success.
 @Override
 public void onNext(List<ByteBuffer> buffers) {
 if (--unfulfilled == (REQUEST_WINDOW / 2)) {
 long req = REQUEST_WINDOW - unfulfilled;
 System.out.println("EndSubscriber request " + req);
+unfulfilled = REQUEST_WINDOW;
 subscription.request(req);
-unfulfilled = REQUEST_WINDOW;
 }
 long currval = counter.get();
 if (currval % 500 == 0) {
-System.out.println("End: " + currval);
+System.out.println("EndSubscriber: " + currval);
 }
 System.out.println("EndSubscriber onNext " + Utils.remaining(buffers));
 for (ByteBuffer buf : buffers) {
 while (buf.hasRemaining()) {

branch	http-client-branch
changeset 55942	8d4770c22b63
parent 55909	583695a0ed6a
child 55947	c4f314605d28