created a reactive socket interface, pointed rxjava2 to the specific version of rxjava2 to get the tests to work, updated Agrona to 0.4.13

Author: robertroeser

build.gradle

@@ -18,9 +18,9 @@ repositories {
 
 dependencies {
 	compile 'org.reactivestreams:reactive-streams:1.0.0.final'
-	compile 'uk.co.real-logic:Agrona:0.4.2'
+	compile 'org.agrona:Agrona:0.4.13'
 
-	testCompile 'io.reactivex:rxjava:2.0.0-DP0-SNAPSHOT'
+	testCompile 'io.reactivex:rxjava:2.0.0-DP0-20151003.214425-143'
 	testCompile 'junit:junit-dep:4.10'
 	testCompile 'org.mockito:mockito-core:1.8.5'
 }

src/main/java/io/reactivesocket/Frame.java

@@ -24,8 +24,8 @@
 import io.reactivesocket.internal.frame.RequestNFrameFlyweight;
 import io.reactivesocket.internal.frame.SetupFrameFlyweight;
 import io.reactivesocket.internal.frame.UnpooledFrame;
-import uk.co.real_logic.agrona.DirectBuffer;
-import uk.co.real_logic.agrona.MutableDirectBuffer;
+import org.agrona.DirectBuffer;
+import org.agrona.MutableDirectBuffer;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;

src/main/java/io/reactivesocket/ReactiveSocket.java

@@ -1,7 +1,108 @@
+/**
+ * Copyright 2015 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.reactivesocket;
 
+import io.reactivesocket.rx.Completable;
+import org.reactivestreams.Publisher;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
 /**
- * Created by rroeser on 3/25/16.
+ * Interface for a connection that supports sending requests and receiving responses
  */
-public interface ReactiveSocket {
+public interface ReactiveSocket extends AutoCloseable {
+    Publisher<Payload> requestResponse(final Payload payload);
+
+    Publisher<Void> fireAndForget(final Payload payload);
+
+    Publisher<Payload> requestStream(final Payload payload);
+
+    Publisher<Payload> requestSubscription(final Payload payload);
+
+    Publisher<Payload> requestChannel(final Publisher<Payload> payloads);
+
+    Publisher<Void> metadataPush(final Payload payload);
+
+    /**
+     * Client check for availability to send request based on lease
+     *
+     * @return 0.0 to 1.0 indicating availability of sending requests
+     */
+    double availability();
+
+    /**
+     * Start protocol processing on the given DuplexConnection.
+     */
+    void start(Completable c);
+
+    /**
+     * Start and block the current thread until startup is finished.
+     *
+     * @throws RuntimeException
+     *             of InterruptedException
+     */
+    default void startAndWait() {
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Throwable> err = new AtomicReference<>();
+        start(new Completable() {
+            @Override
+            public void success() {
+                latch.countDown();
+            }
+
+            @Override
+            public void error(Throwable e) {
+                latch.countDown();
+            }
+        });
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+        if (err.get() != null) {
+            throw new RuntimeException(err.get());
+        }
+    }
+
+    /**
+     * Invoked when Requester is ready. Non-null exception if error. Null if success.
+     *
+     * @param c
+     */
+    void onRequestReady(Consumer<Throwable> c);
+
+    /**
+     * Invoked when Requester is ready with success or fail.
+     *
+     * @param c
+     */
+    void onRequestReady(Completable c);
+
+    /**
+     * Server granting new lease information to client
+     *
+     * Initial lease semantics are that server waits for periodic granting of leases by server side.
+     *
+     * @param ttl
+     * @param numberOfRequests
+     */
+    void sendLease(int ttl, int numberOfRequests);
+
+    void shutdown();
 }

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

@@ -1,3 +1,18 @@
+/**
+ * Copyright 2015 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.reactivesocket;
 
 import org.reactivestreams.Publisher;

src/main/java/io/reactivesocket/ReactiveSocketImpl.java

@@ -15,17 +15,6 @@
  */
 package io.reactivesocket;
 
-import java.io.IOException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
-
-import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-
 import io.reactivesocket.internal.Requester;
 import io.reactivesocket.internal.Responder;
 import io.reactivesocket.internal.rx.CompositeCompletable;
@@ -34,16 +23,21 @@
 import io.reactivesocket.rx.Disposable;
 import io.reactivesocket.rx.Observable;
 import io.reactivesocket.rx.Observer;
-import uk.co.real_logic.agrona.BitUtil;
+import org.agrona.BitUtil;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 
 import static io.reactivesocket.LeaseGovernor.NULL_LEASE_GOVERNOR;
 
 /**
- * Interface for a connection that supports sending requests and receiving responses
- *
- * Created by servers for connections Created on demand for clients
+ * An implementation of {@link ReactiveSocket}
  */
-public class ReactiveSocket implements AutoCloseable {
+public class ReactiveSocketImpl implements ReactiveSocket {
     private static final RequestHandler EMPTY_HANDLER = new RequestHandler.Builder().build();
 
     private static final Consumer<Throwable> DEFAULT_ERROR_STREAM = t -> {
@@ -62,7 +56,7 @@ public class ReactiveSocket implements AutoCloseable {
     private final ConnectionSetupHandler responderConnectionHandler;
     private final LeaseGovernor leaseGovernor;
 
-    private ReactiveSocket(
+    private ReactiveSocketImpl(
         DuplexConnection connection,
         boolean isServer,
         ConnectionSetupPayload serverRequestorSetupPayload,
@@ -118,7 +112,7 @@ public static ReactiveSocket fromClientConnection(
         }
         final RequestHandler h = handler != null ? handler : EMPTY_HANDLER;
         Consumer<Throwable> es = errorStream != null ? errorStream : DEFAULT_ERROR_STREAM;
-        return new ReactiveSocket(connection, false, setup, h, null, NULL_LEASE_GOVERNOR, es);
+        return new ReactiveSocketImpl(connection, false, setup, h, null, NULL_LEASE_GOVERNOR, es);
     }
 
     /**
@@ -178,7 +172,7 @@ public static ReactiveSocket fromServerConnection(
         LeaseGovernor leaseGovernor,
         Consumer<Throwable> errorConsumer
     ) {
-        return new ReactiveSocket(connection, true, null, null, connectionHandler,
+        return new ReactiveSocketImpl(connection, true, null, null, connectionHandler,
             leaseGovernor, errorConsumer);
     }
 
@@ -192,31 +186,37 @@ public static ReactiveSocket fromServerConnection(
     /**
      * Initiate a request response exchange
      */
+    @Override
     public Publisher<Payload> requestResponse(final Payload payload) {
         assertRequester();
         return requester.requestResponse(payload);
     }
 
+    @Override
     public Publisher<Void> fireAndForget(final Payload payload) {
         assertRequester();
         return requester.fireAndForget(payload);
     }
 
+    @Override
     public Publisher<Payload> requestStream(final Payload payload) {
         assertRequester();
         return requester.requestStream(payload);
     }
 
+    @Override
     public Publisher<Payload> requestSubscription(final Payload payload) {
         assertRequester();
         return requester.requestSubscription(payload);
     }
 
+    @Override
     public Publisher<Payload> requestChannel(final Publisher<Payload> payloads) {
         assertRequester();
         return requester.requestChannel(payloads);
     }
 
+    @Override
     public Publisher<Void> metadataPush(final Payload payload) {
         assertRequester();
         return requester.metadataPush(payload);
@@ -239,33 +239,20 @@ private void assertRequester() {
         }
     }
 
-    /**
-     * Client check for availability to send request based on lease
-     *
-     * @return 0.0 to 1.0 indicating availability of sending requests
-     */
+    @Override
     public double availability() {
         // TODO: can happen in either direction
         assertRequester();
         return requester.availability();
     }
 
-    /**
-     * Server granting new lease information to client
-     *
-     * Initial lease semantics are that server waits for periodic granting of leases by server side.
-     *
-     * @param ttl
-     * @param numberOfRequests
-     */
+    @Override
     public void sendLease(int ttl, int numberOfRequests) {
         // TODO: can happen in either direction
         responder.sendLease(ttl, numberOfRequests);
     }
 
-    /**
-     * Start protocol processing on the given DuplexConnection.
-     */
+    @Override
     public final void start(Completable c) {
         if (isServer) {
             responder = Responder.createServerResponder(
@@ -345,20 +332,12 @@ public void error(Throwable e) {
 
     private final CompositeCompletable requesterReady = new CompositeCompletable();
 
-    /**
-     * Invoked when Requester is ready with success or fail.
-     *
-     * @param c
-     */
+    @Override
     public final void onRequestReady(Completable c) {
         requesterReady.add(c);
     }
 
-    /**
-     * Invoked when Requester is ready. Non-null exception if error. Null if success.
-     *
-     * @param c
-     */
+    @Override
     public final void onRequestReady(Consumer<Throwable> c) {
         requesterReady.add(new Completable() {
             @Override
@@ -458,36 +437,6 @@ public void addOutput(Frame f, Completable callback) {
 
     };
 
-    /**
-     * Start and block the current thread until startup is finished.
-     *
-     * @throws RuntimeException
-     *             of InterruptedException
-     */
-    public final void startAndWait() {
-        CountDownLatch latch = new CountDownLatch(1);
-        AtomicReference<Throwable> err = new AtomicReference<>();
-        start(new Completable() {
-            @Override
-            public void success() {
-                latch.countDown();
-            }
-
-            @Override
-            public void error(Throwable e) {
-                latch.countDown();
-            }
-        });
-        try {
-            latch.await();
-        } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-        }
-        if (err.get() != null) {
-            throw new RuntimeException(err.get());
-        }
-    }
-
     @Override
     public void close() throws Exception {
         connection.close();
@@ -500,6 +449,7 @@ public void close() throws Exception {
         }
     }
 
+    @Override
     public void shutdown() {
         try {
             close();

src/main/java/io/reactivesocket/internal/Requester.java

@@ -44,7 +44,7 @@
 import io.reactivesocket.rx.Completable;
 import io.reactivesocket.rx.Disposable;
 import io.reactivesocket.rx.Observer;
-import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Int2ObjectHashMap;
 
 /**
  * Protocol implementation abstracted over a {@link DuplexConnection}.