embedded Aeron MediaDriver in the client (#6)

* aeron client can run embededded media driver so you don't have to run the external driver

* added toString methods for duplex connections for debugging and access log purposes

Author: robertroeser

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/AeronClientDuplexConnection.java

@@ -126,5 +126,15 @@ public CopyOnWriteArrayList<Observer<Frame>> getSubjects() {
         return subjects;
     }
 
+    public String toString() {
+        if (publication == null) {
+            return  getClass().getName() + ":publication=null";
+        }
+
+        return getClass().getName() + ":publication=[" +
+            "channel=" + publication.channel() + "," +
+            "streamId=" + publication.streamId() + "," +
+            "sessionId=" + publication.sessionId() + "]";
 
+    }
 }

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/client/ClientAeronManager.java

@@ -15,14 +15,19 @@
  */
 package io.reactivesocket.aeron.client;
 
-import io.reactivesocket.aeron.internal.Loggable;
-import rx.Scheduler;
-import rx.schedulers.Schedulers;
 import io.aeron.Aeron;
 import io.aeron.FragmentAssembler;
 import io.aeron.Image;
 import io.aeron.Subscription;
+import io.aeron.driver.MediaDriver;
+import io.aeron.driver.ThreadingMode;
 import io.aeron.logbuffer.FragmentHandler;
+import io.reactivesocket.aeron.internal.Constants;
+import io.reactivesocket.aeron.internal.Loggable;
+import org.agrona.concurrent.BackoffIdleStrategy;
+import org.agrona.concurrent.SleepingIdleStrategy;
+import rx.Scheduler;
+import rx.schedulers.Schedulers;
 
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
@@ -33,6 +38,24 @@
 public class ClientAeronManager implements Loggable {
     private static final ClientAeronManager INSTANCE = new ClientAeronManager();
 
+    /**
+     * Enables running the client with an embedded Aeron {@link MediaDriver} so you don't have to run
+     * the driver in a separate process. To enable this option you need to set the reactivesocket.aeron.clientEmbeddedDriver
+     * to true
+     */
+    static {
+       if (Constants.CLIENT_EMBEDDED_AERON_DRIVER) {
+           System.out.println("+++ Launching embedded media driver");
+           final MediaDriver.Context context = new MediaDriver.Context();
+           context.dirsDeleteOnStart(true);
+           context.threadingMode(ThreadingMode.SHARED_NETWORK);
+           context.conductorIdleStrategy(new SleepingIdleStrategy(TimeUnit.MILLISECONDS.toNanos(10)));
+           context.senderIdleStrategy(new BackoffIdleStrategy(5, 10, 100, 1000));
+           context.receiverIdleStrategy(new BackoffIdleStrategy(5, 10, 100, 1000));
+           MediaDriver.launch(context);
+       }
+    }
+
     private final CopyOnWriteArrayList<ClientAction> clientActions;
 
     private final CopyOnWriteArrayList<SubscriptionGroup> subscriptionGroups;

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/internal/Constants.java

@@ -39,6 +39,7 @@ public final class Constants {
     public static final int SERVER_TIMER_WHEEL_TICK_DURATION_MS = 10;
     public static final int SERVER_TIMER_WHEEL_BUCKETS = 128;
     public static final int DEFAULT_OFFER_TO_AERON_TIMEOUT_MS = 30_000;
+    public static final boolean CLIENT_EMBEDDED_AERON_DRIVER = Boolean.getBoolean("reactivesocket.aeron.clientEmbeddedDriver");
 
     static {
         String idlStrategy = System.getProperty("idleStrategy");

reactivesocket-aeron/src/main/java/io/reactivesocket/aeron/server/AeronServerDuplexConnection.java

@@ -107,4 +107,16 @@ public void close() {
             publication.close();
         } catch (Throwable t) {}
     }
+
+    public String toString() {
+        if (publication == null) {
+            return  getClass().getName() + ":publication=null";
+        }
+
+        return getClass().getName() + ":publication=[" +
+            "channel=" + publication.channel() + "," +
+            "streamId=" + publication.streamId() + "," +
+            "sessionId=" + publication.sessionId() + "]";
+
+    }
 }

reactivesocket-jsr-356/src/main/java/io/reactivesocket/javax/websocket/WebSocketDuplexConnection.java

@@ -83,4 +83,13 @@ public void onSubscribe(Subscription s) {
     public void close() throws IOException {
         session.close();
     }
+
+    public String toString() {
+        if (session == null) {
+            return  getClass().getName() + ":session=null";
+        }
+
+        return getClass().getName() + ":session=[" + session.toString() + "]";
+
+    }
 }

reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/client/ClientTcpDuplexConnection.java

@@ -151,4 +151,20 @@ public void onComplete() {
     public void close() throws IOException {
         channel.close();
     }
+
+    public String toString() {
+        if (channel == null) {
+            return  getClass().getName() + ":channel=null";
+        }
+
+        return getClass().getName() + ":channel=[" +
+            "remoteAddress=" + channel.remoteAddress() + "," +
+            "isActive=" + channel.isActive() + "," +
+            "isOpen=" + channel.isOpen() + "," +
+            "isRegistered=" + channel.isRegistered() + "," +
+            "isWritable=" + channel.isWritable() + "," +
+            "channelId=" + channel.id().asLongText() +
+            "]";
+
+    }
 }

reactivesocket-netty/src/main/java/io/reactivesocket/netty/tcp/server/ServerTcpDuplexConnection.java

@@ -17,6 +17,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.reactivesocket.DuplexConnection;
@@ -97,4 +98,21 @@ public void onComplete() {
     public void close() throws IOException {
 
     }
+
+    public String toString() {
+        if (ctx ==null || ctx.channel() == null) {
+            return  getClass().getName() + ":channel=null";
+        }
+
+        Channel channel = ctx.channel();
+        return getClass().getName() + ":channel=[" +
+            "remoteAddress=" + channel.remoteAddress() + "," +
+            "isActive=" + channel.isActive() + "," +
+            "isOpen=" + channel.isOpen() + "," +
+            "isRegistered=" + channel.isRegistered() + "," +
+            "isWritable=" + channel.isWritable() + "," +
+            "channelId=" + channel.id().asLongText() +
+            "]";
+
+    }
 }

reactivesocket-netty/src/main/java/io/reactivesocket/netty/websocket/client/ClientWebSocketDuplexConnection.java

@@ -166,4 +166,20 @@ public void onComplete() {
     public void close() throws IOException {
         channel.close();
     }
+
+    public String toString() {
+        if (channel == null) {
+            return  getClass().getName() + ":channel=null";
+        }
+
+        return getClass().getName() + ":channel=[" +
+            "remoteAddress=" + channel.remoteAddress() + "," +
+            "isActive=" + channel.isActive() + "," +
+            "isOpen=" + channel.isOpen() + "," +
+            "isRegistered=" + channel.isRegistered() + "," +
+            "isWritable=" + channel.isWritable() + "," +
+            "channelId=" + channel.id().asLongText() +
+            "]";
+
+    }
 }

reactivesocket-netty/src/main/java/io/reactivesocket/netty/websocket/server/ServerWebSocketDuplexConnection.java

@@ -17,6 +17,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
@@ -99,4 +100,21 @@ public void onComplete() {
     public void close() throws IOException {
 
     }
+
+    public String toString() {
+        if (ctx ==null || ctx.channel() == null) {
+            return  getClass().getName() + ":channel=null";
+        }
+
+        Channel channel = ctx.channel();
+        return getClass().getName() + ":channel=[" +
+            "remoteAddress=" + channel.remoteAddress() + "," +
+            "isActive=" + channel.isActive() + "," +
+            "isOpen=" + channel.isOpen() + "," +
+            "isRegistered=" + channel.isRegistered() + "," +
+            "isWritable=" + channel.isWritable() + "," +
+            "channelId=" + channel.id().asLongText() +
+            "]";
+
+    }
 }