updated ReactiveSocketFactory

Author: robertroeser

src/main/java/io/reactivesocket/ReactiveSocketFactory.java

@@ -15,12 +15,111 @@
  */
 package io.reactivesocket;
 
+import io.reactivesocket.internal.rx.EmptySubscription;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 
-import java.net.SocketAddress;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 @FunctionalInterface
-public interface ReactiveSocketFactory {
-    Publisher<ReactiveSocket> call(SocketAddress address, long timeout, TimeUnit timeUnit);
+public interface ReactiveSocketFactory<T, R extends ReactiveSocket> {
+
+    Publisher<R> call(T t);
+
+    /**
+     * Gets a socket in a blocking manner
+     * @param t configuration to create the reactive socket
+     * @return blocks on create the socket
+     */
+    default R callAndWait(T t) {
+        AtomicReference<R> reference = new AtomicReference<>();
+        AtomicReference<Throwable> error = new AtomicReference<>();
+        CountDownLatch latch = new CountDownLatch(1);
+
+        call(t)
+            .subscribe(new Subscriber<R>() {
+                @Override
+                public void onSubscribe(Subscription s) {
+                    s.request(1);
+                }
+
+                @Override
+                public void onNext(R reactiveSocket) {
+                    reference.set(reactiveSocket);
+                }
+
+                @Override
+                public void onError(Throwable t) {
+                    error.set(t);
+                    latch.countDown();
+                }
+
+                @Override
+                public void onComplete() {
+                    latch.countDown();
+                }
+            });
+
+        if (error.get() != null) {
+            throw new RuntimeException(error.get());
+        } else {
+            return reference.get();
+        }
+    }
+
+    /**
+     *
+     * @param t the configuration used to create the reactive socket
+     * @param timeout timeout
+     * @param timeUnit timeout units
+     * @param executorService ScheduledExecutorService to schedule the timeout on
+     * @return
+     */
+    default Publisher<R> call(T t, long timeout, TimeUnit timeUnit, ScheduledExecutorService executorService) {
+        Publisher<R> reactiveSocketPublisher = subscriber -> {
+            AtomicBoolean complete = new AtomicBoolean();
+            subscriber.onSubscribe(EmptySubscription.INSTANCE);
+            call(t)
+                .subscribe(new Subscriber<R>() {
+                    @Override
+                    public void onSubscribe(Subscription s) {
+                        s.request(1);
+                    }
+
+                    @Override
+                    public void onNext(R reactiveSocket) {
+                        subscriber.onNext(reactiveSocket);
+                    }
+
+                    @Override
+                    public void onError(Throwable t) {
+                        subscriber.onError(t);
+                    }
+
+                    @Override
+                    public void onComplete() {
+                        if (!complete.get()) {
+                            complete.set(true);
+                            subscriber.onComplete();
+                        }
+                    }
+                });
+
+            executorService.schedule(() -> {
+                if (!complete.get()) {
+                    complete.set(true);
+                    subscriber.onError(new TimeoutException());
+                }
+            }, timeout, timeUnit);
+        };
+
+        return reactiveSocketPublisher;
+    }
+
 }

src/main/java/io/reactivesocket/ReactiveSocketSocketAddressFactory.java

@@ -0,0 +1,8 @@
+package io.reactivesocket;
+
+import java.net.SocketAddress;
+
+@FunctionalInterface
+public interface ReactiveSocketSocketAddressFactory<R extends ReactiveSocket> extends ReactiveSocketFactory<SocketAddress, R> {
+
+}