fixed code that establishes a connection:

Author: robertroeser

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

@@ -68,23 +68,26 @@ public Publisher<Void> write(Publisher<Frame> o) {
         return RxReactiveStreams.toPublisher(req);
     }
 
-    void establishConnection() {
+    void ackEstablishConnection(int ackSessionId) {
         final long start = System.nanoTime();
         final int sessionId = publication.sessionId();
         final BufferClaim bufferClaim = bufferClaims.get();
 
+        System.out.print("Acking establish connection for session id => " + ackSessionId);
+
         for (;;) {
             final long current = System.nanoTime();
             if (current - start > TimeUnit.SECONDS.toNanos(30)) {
                 throw new RuntimeException("Timed out waiting to establish connection for session id => " + sessionId);
             }
 
-            final long offer = publication.tryClaim(BitUtil.SIZE_OF_INT, bufferClaim);
+            final long offer = publication.tryClaim(2 * BitUtil.SIZE_OF_INT, bufferClaim);
             if (offer >= 0) {
                 try {
                     final MutableDirectBuffer buffer = bufferClaim.buffer();
                     final int offeset = bufferClaim.offset();
                     buffer.putInt(offeset, MessageType.ESTABLISH_CONNECTION_RESPONSE.getEncodedType());
+                    buffer.putInt(offeset + BitUtil.SIZE_OF_INT, ackSessionId);
                 } finally {
                     bufferClaim.commit();
                 }

src/main/java/io/reactivesocket/aeron/ReactiveSocketAeronServer.java

@@ -18,6 +18,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
 
 import static io.reactivesocket.aeron.Constants.CLIENT_STREAM_ID;
 import static io.reactivesocket.aeron.Constants.SERVER_STREAM_ID;
@@ -30,7 +31,7 @@ public class ReactiveSocketAeronServer implements AutoCloseable {
 
     private final int port;
 
-    private final Int2ObjectHashMap<AeronServerDuplexConnection> connections;
+    private volatile Int2ObjectHashMap<AeronServerDuplexConnection> connections;
 
     private final Scheduler.Worker worker;
 
@@ -78,7 +79,7 @@ void poll(FragmentAssembler fragmentAssembler) {
     void fragmentHandler(DirectBuffer buffer, int offset, int length, Header header) {
         final int sessionId = header.sessionId();
 
-        int messageTypeInt = buffer.getInt(0);
+        int messageTypeInt = buffer.getInt(offset);
         MessageType type = MessageType.from(messageTypeInt);
 
         if (MessageType.FRAME == type) {
@@ -88,26 +89,39 @@ void fragmentHandler(DirectBuffer buffer, int offset, int length, Header header)
             if (connection != null) {
                 final PublishSubject<Frame> subject = connection.getSubject();
                 ByteBuffer bytes = ByteBuffer.allocate(buffer.capacity());
-                buffer.getBytes(BitUtil.SIZE_OF_INT, bytes, buffer.capacity());
+                buffer.getBytes(BitUtil.SIZE_OF_INT + offset, bytes, buffer.capacity());
                 final Frame frame = Frame.from(bytes);
                 subject.onNext(frame);
             }
         } else if (MessageType.ESTABLISH_CONNECTION_REQUEST == type) {
-            AeronServerDuplexConnection connection = connections.get(sessionId);
-            connection.establishConnection();
+            final long start = System.nanoTime();
+            AeronServerDuplexConnection connection = null;
+            System.out.println("Looking a connection to ack establish connection for session id => " + sessionId);
+            while (connection == null) {
+                final long current = System.nanoTime();
+
+                if (current - start > TimeUnit.SECONDS.toNanos(30)) {
+                    throw new RuntimeException("unable to find connection to ack establish connection for session id => " + sessionId);
+                }
+
+                connection = connections.get(sessionId);
+            }
+            System.out.println("Found a connection to ack establish connection for session id => " + sessionId);
+            connection.ackEstablishConnection(sessionId);
         }
 
     }
 
     void newImageHandler(Image image, String channel, int streamId, int sessionId, long joiningPosition, String sourceIdentity) {
-        System.out.println(String.format("Handling new image for session id => %d and stream id => %d", streamId, sessionId));
         if (SERVER_STREAM_ID == streamId) {
+            System.out.println(String.format("Handling new image for session id => %d and stream id => %d", streamId, sessionId));
             final AeronServerDuplexConnection connection = connections.computeIfAbsent(sessionId, (_s) -> {
                 final String responseChannel = "udp://" + sourceIdentity.substring(0, sourceIdentity.indexOf(':')) + ":" + port;
                 Publication publication = aeron.addPublication(responseChannel, CLIENT_STREAM_ID);
                 System.out.println(String.format("Creating new connection for responseChannel => %s, streamId => %d, and sessionId => %d", responseChannel, streamId, sessionId));
                 return new AeronServerDuplexConnection(publication);
             });
+            System.out.println("Accepting ReactiveSocket connection");
             rsServerProtocol.acceptConnection(connection);
         } else {
             System.out.println("Unsupported stream id " + streamId);

src/main/java/io/reactivesocket/aeron/ReactivesocketAeronClient.java

@@ -124,7 +124,7 @@ public static ReactivesocketAeronClient create(String host) {
     }
 
     void fragmentHandler(DirectBuffer buffer, int offset, int length, Header header) {
-        int messageTypeInt = buffer.getInt(0);
+        int messageTypeInt = buffer.getInt(offset);
         MessageType messageType = MessageType.from(messageTypeInt);
         if (messageType == MessageType.FRAME) {
             final PublishSubject<Frame> subject = subjects.get(header.sessionId());
@@ -133,7 +133,9 @@ void fragmentHandler(DirectBuffer buffer, int offset, int length, Header header)
             final Frame frame = Frame.from(bytes);
             subject.onNext(frame);
         } else if (messageType == MessageType.ESTABLISH_CONNECTION_RESPONSE) {
-            CountDownLatch latch = establishConnectionLatches.get(header.sessionId());
+            int ackSessionId = buffer.getInt(offset + BitUtil.SIZE_OF_INT);
+            System.out.println(String.format("Received establish connection ack for session id => %d", ackSessionId));
+            CountDownLatch latch = establishConnectionLatches.get(ackSessionId);
             latch.countDown();
         } else {
             System.out.println("Unknow message type => " + messageTypeInt);
@@ -150,7 +152,7 @@ void poll(FragmentAssembler fragmentAssembler, Subscription subscription, Schedu
     }
 
     /**
-     * Establishes a connection between the client and server. Waits for 30 seconds before throwing a exception.
+     * Establishes a connection between the client and server. Waits for 39 seconds before throwing a exception.
      */
     void establishConnection(final Publication publication, final int sessionId) {
         try {
@@ -173,7 +175,7 @@ void establishConnection(final Publication publication, final int sessionId) {
                     offer = publication.offer(buffer);
                 }
 
-                if (latch.getCount() > 0) {
+                if (latch.getCount() == 0) {
                     break;
                 }
             }