@@ -0,0 +1,2620 @@
+package io.reactivesocket.util;%0A%0Aimport io.reactivesocket.ReactiveSocket;%0Aimport io.reactivesocket.rx.Completable;%0Aimport org.reactivestreams.Publisher;%0Aimport org.reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription;%0A%0Aimport java.util.ArrayList;%0Aimport java.util.List;%0Aimport java.util.concurrent.*;%0A%0Apublic class Unsafe %7B%0A    public static ReactiveSocket startAndWait(ReactiveSocket rsc) throws InterruptedException %7B%0A        CountDownLatch latch = new CountDownLatch(1);%0A        Completable completable = new Completable() %7B%0A            @Override%0A            public void success() %7B%0A                latch.countDown();%0A            %7D%0A%0A            @Override%0A            public void error(Throwable e) %7B%0A                latch.countDown();%0A            %7D%0A        %7D;%0A        rsc.start(completable);%0A        latch.await();%0A//        awaitAvailability(rsc);%0A%0A        return rsc;%0A    %7D%0A%0A    public static ReactiveSocket awaitAvailability(ReactiveSocket rsc) throws InterruptedException %7B%0A        long waiting = 1L;%0A        while (rsc.availability() == 0.0) %7B%0A            Thread.sleep(waiting);%0A            waiting = Math.max(waiting * 2, 1000L);%0A        %7D%0A        return rsc;%0A    %7D%0A%0A    public static %3CT%3E T blockingSingleWait(Publisher%3CT%3E publisher, long timeout, TimeUnit unit)%0A        throws InterruptedException, ExecutionException, TimeoutException %7B%0A        return toSingleFuture(publisher).get(timeout, unit);%0A    %7D%0A%0A    public static %3CT%3E List%3CT%3E blockingWait(Publisher%3CT%3E publisher, long timeout, TimeUnit unit)%0A        throws InterruptedException, ExecutionException, TimeoutException %7B%0A        return toFuture(publisher).get(timeout, unit);%0A    %7D%0A%0A    public static %3CT%3E CompletableFuture%3CT%3E toSingleFuture(Publisher%3CT%3E publisher) %7B%0A        return toFuture(publisher).thenApply(list -%3E list.get(0));%0A    %7D%0A%0A    public static %3CT%3E CompletableFuture%3CList%3CT%3E%3E toFuture(Publisher%3CT%3E publisher) %7B%0A        CompletableFuture%3CList%3CT%3E%3E future = new CompletableFuture%3C%3E();%0A%0A        publisher.subscribe(new Subscriber%3CT%3E() %7B%0A            private List%3CT%3E buffer = new ArrayList%3CT%3E();%0A%0A            @Override%0A            public void onSubscribe(Subscription s) %7B%0A                s.request(Long.MAX_VALUE);%0A            %7D%0A%0A            @Override%0A            public void onNext(T t) %7B%0A                buffer.add(t);%0A            %7D%0A%0A            @Override%0A            public void onError(Throwable t) %7B%0A                future.completeExceptionally(t);%0A            %7D%0A%0A            @Override%0A            public void onComplete() %7B%0A                future.complete(buffer);%0A            %7D%0A        %7D);%0A%0A        return future;%0A    %7D%0A%7D%0A

Add missing file Help Unsafe object