83         FlowTube server = new SSLTube(createSSLEngine(false),

    94 //        FlowTube server = new SSLTube(createSSLEngine(false),

    84                                       sslExecutor,

    95 //                                      sslExecutor,

    85                                       new EchoTube(16));

    96 //                                      new EchoTube(16));

    97         SSLLoopbackSubscriber server =

    98                 new SSLLoopbackSubscriber((new SimpleSSLContext()).get(), sslExecutor);

    99         server.start();

   100 

   116     private static final class EchoTube implements FlowTube {

   131     static class SSLLoopbackSubscriber implements FlowTube {

   117 

   132         private final BlockingQueue<ByteBuffer> buffer;

   118         private final static Object EOF = new Object();

   133         private final Socket clientSock;

   119         private final Executor executor = Executors.newSingleThreadExecutor();

   134         private final SSLSocket serverSock;

   120 

   135         private final Thread thread1, thread2, thread3;

   121         private final Queue<Object> queue = new ConcurrentLinkedQueue<>();

   136         private volatile Flow.Subscription clientSubscription;

   122         private final int maxQueueSize;

   137         private final SubmissionPublisher<List<ByteBuffer>> publisher;

   123         private final SequentialScheduler processingScheduler =

   138 

   124                 new SequentialScheduler(createProcessingTask());

   139         SSLLoopbackSubscriber(SSLContext ctx, ExecutorService exec) throws IOException {

   125 

   140             SSLServerSocketFactory fac = ctx.getServerSocketFactory();

   126         /* Writing into this tube */

   141             SSLServerSocket serv = (SSLServerSocket) fac.createServerSocket(0);

   127         private long unfulfilled;

   142             SSLParameters params = serv.getSSLParameters();

   128         private Flow.Subscription subscription;

   143             params.setApplicationProtocols(new String[]{"proto2"});

   129 

   144             serv.setSSLParameters(params);

   130         /* Reading from this tube */

   145 

   131         private final Demand demand = new Demand();

   146 

   132         private final AtomicBoolean cancelled = new AtomicBoolean();

   147             int serverPort = serv.getLocalPort();

   133         private Flow.Subscriber<? super List<ByteBuffer>> subscriber;

   148             clientSock = new Socket("127.0.0.1", serverPort);

   134 

   149             serverSock = (SSLSocket) serv.accept();

   135         private EchoTube(int maxBufferSize) {

   150             this.buffer = new LinkedBlockingQueue<>();

   136             if (maxBufferSize < 1)

   151             thread1 = new Thread(this::clientWriter, "clientWriter");

   137                 throw new IllegalArgumentException();

   152             thread2 = new Thread(this::serverLoopback, "serverLoopback");

   138             this.maxQueueSize = maxBufferSize;

   153             thread3 = new Thread(this::clientReader, "clientReader");

   139         }

   154             publisher = new SubmissionPublisher<>(exec, Flow.defaultBufferSize(),

   140 

   155                     this::handlePublisherException);

   141         @Override

   156             SSLFlowDelegate.Monitor.add(this::monitor);

   142         public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {

   157         }

   143             this.subscriber = subscriber;

   158 

   144             this.subscriber.onSubscribe(new InternalSubscription());

   159         public void start() {

   160             thread1.start();

   161             thread2.start();

   162             thread3.start();

   163         }

   164 

   165         private void handlePublisherException(Object o, Throwable t) {

   166             System.out.println("Loopback Publisher exception");

   167             t.printStackTrace(System.out);

   168         }

   169 

   170         private final AtomicInteger readCount = new AtomicInteger();

   171 

   172         // reads off the SSLSocket the data from the "server"

   173         private void clientReader() {

   174             try {

   175                 InputStream is = clientSock.getInputStream();

   176                 final int bufsize = FlowTest.randomRange(512, 16 * 1024);

   177                 System.out.println("clientReader: bufsize = " + bufsize);

   178                 while (true) {

   179                     byte[] buf = new byte[bufsize];

   180                     int n = is.read(buf);

   181                     if (n == -1) {

   182                         System.out.println("clientReader close: read "

   183                                 + readCount.get() + " bytes");

   184                         publisher.close();

   185                         sleep(2000);

   186                         Utils.close(is, clientSock);

   187                         return;

   188                     }

   189                     ByteBuffer bb = ByteBuffer.wrap(buf, 0, n);

   190                     readCount.addAndGet(n);

   191                     publisher.submit(List.of(bb));

   192                 }

   193             } catch (Throwable e) {

   194                 e.printStackTrace();

   195                 Utils.close(clientSock);

   196             }

   197         }

   198 

   199         // writes the encrypted data from SSLFLowDelegate to the j.n.Socket

   200         // which is connected to the SSLSocket emulating a server.

   201         private void clientWriter() {

   202             long nbytes = 0;

   203             try {

   204                 OutputStream os =

   205                         new BufferedOutputStream(clientSock.getOutputStream());

   206 

   207                 while (true) {

   208                     ByteBuffer buf = buffer.take();

   209                     if (buf == FlowTest.SENTINEL) {

   210                         // finished

   211                         //Utils.sleep(2000);

   212                         System.out.println("clientWriter close: " + nbytes + " written");

   213                         clientSock.shutdownOutput();

   214                         System.out.println("clientWriter close return");

   215                         return;

   216                     }

   217                     int len = buf.remaining();

   218                     int written = writeToStream(os, buf);

   219                     assert len == written;

   220                     nbytes += len;

   221                     assert !buf.hasRemaining()

   222                             : "buffer has " + buf.remaining() + " bytes left";

   223                     clientSubscription.request(1);

   224                 }

   225             } catch (Throwable e) {

   226                 e.printStackTrace();

   227             }

   228         }

   229 

   230         private int writeToStream(OutputStream os, ByteBuffer buf) throws IOException {

   231             byte[] b = buf.array();

   232             int offset = buf.arrayOffset() + buf.position();

   233             int n = buf.limit() - buf.position();

   234             os.write(b, offset, n);

   235             buf.position(buf.limit());

   236             os.flush();

   237             return n;

   238         }

   239 

   240         private final AtomicInteger loopCount = new AtomicInteger();

   241 

   242         public String monitor() {

   243             return "serverLoopback: loopcount = " + loopCount.toString()

   244                     + " clientRead: count = " + readCount.toString();

   245         }

   246 

   247         // thread2

   248         private void serverLoopback() {

   249             try {

   250                 InputStream is = serverSock.getInputStream();

   251                 OutputStream os = serverSock.getOutputStream();

   252                 final int bufsize = FlowTest.randomRange(512, 16 * 1024);

   253                 System.out.println("serverLoopback: bufsize = " + bufsize);

   254                 byte[] bb = new byte[bufsize];

   255                 while (true) {

   256                     int n = is.read(bb);

   257                     if (n == -1) {

   258                         sleep(2000);

   259                         is.close();

   260                         os.close();

   261                         serverSock.close();

   262                         return;

   263                     }

   264                     os.write(bb, 0, n);

   265                     os.flush();

   266                     loopCount.addAndGet(n);

   267                 }

   268             } catch (Throwable e) {

   269                 e.printStackTrace();

   270             }

   271         }

   272 

   273 

   274         /**

   275          * This needs to be called before the chain is subscribed. It can't be

   276          * supplied in the constructor.

   277          */

   278         public void setReturnSubscriber(Flow.Subscriber<List<ByteBuffer>> returnSubscriber) {

   279             publisher.subscribe(returnSubscriber);

   149             unfulfilled = maxQueueSize;

   284             clientSubscription = subscription;

   150             (this.subscription = subscription).request(maxQueueSize);

   285             clientSubscription.request(5);

   155             if (--unfulfilled == (maxQueueSize / 2)) {

   290             try {

   156                 subscription.request(maxQueueSize - unfulfilled);

   291                 for (ByteBuffer b : item)

   157                 unfulfilled = maxQueueSize;

   292                     buffer.put(b);

   158             }

   293             } catch (InterruptedException e) {

   159             queue.add(item);

   294                 e.printStackTrace();

   160             processingScheduler.deferOrSchedule(executor);

   295                 Utils.close(clientSock);

   296             }

   165             queue.add(throwable);

   301             throwable.printStackTrace();

   166             processingScheduler.deferOrSchedule(executor);

   302             Utils.close(clientSock);

   171             queue.add(EOF);

   307             try {

   172             processingScheduler.deferOrSchedule(executor);

   308                 buffer.put(FlowTest.SENTINEL);

   309             } catch (InterruptedException e) {

   310                 e.printStackTrace();

   311                 Utils.close(clientSock);

   312             }

   180         private class InternalSubscription implements Flow.Subscription {

   320         @Override

   181 

   321         public void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) {

   182             @Override

   322             publisher.subscribe(subscriber);

   183             public void request(long n) {

   184                 if (n <= 0) {

   185                     throw new InternalError();

   186                 }

   187                 demand.increase(n);

   188                 processingScheduler.runOrSchedule();

   189             }

   190 

   191             @Override

   192             public void cancel() {

   193                 cancelled.set(true);

   194             }

   195         }

   196 

   197         private SequentialScheduler.RestartableTask createProcessingTask() {

   198             return new SequentialScheduler.CompleteRestartableTask() {

   199 

   200                 @Override

   201                 protected void run() {

   202                     while (!cancelled.get()) {

   203                         Object item = queue.peek();

   204                         if (item == null)

   205                             return;

   206                         try {

   207                             if (item instanceof List) {

   208                                 if (!demand.tryDecrement())

   209                                     return;

   210                                 @SuppressWarnings("unchecked")

   211                                 List<ByteBuffer> bytes = (List<ByteBuffer>) item;

   212                                 subscriber.onNext(bytes);

   213                             } else if (item instanceof Throwable) {

   214                                 cancelled.set(true);

   215                                 subscriber.onError((Throwable) item);

   216                             } else if (item == EOF) {

   217                                 cancelled.set(true);

   218                                 subscriber.onComplete();

   219                             } else {

   220                                 throw new InternalError(String.valueOf(item));

   221                             }

   222                         } finally {

   223                             Object removed = queue.remove();

   224                             assert removed == item;

   225                         }

   226                     }

   227                 }

   228             };

   272                 subscription.request(REQUEST_WINDOW - unfulfilled);

   519                 long req = REQUEST_WINDOW - unfulfilled;

   520                 System.out.println("EndSubscriber request " + req);

   521                 subscription.request(req);