provides fixes for WrappedDirectBufferByteBuf

uncomments largePayload tests

Signed-off-by: Oleh Dokuka <shadowgun@i.ua>

Author: OlegDokuka

rsocket-transport-aeron/src/main/java/io/rsocket/transport/aeron/AeronClientTransport.java

@@ -117,7 +117,9 @@ public Mono<DuplexConnection> connect() {
 
                 return Mono.error(
                     new TimeoutException(
-                        "Timeout on send SetupFrame { connection: [{}]; stream: [{}]; channel: [{}] }"));
+                        String.format(
+                            "Timeout on send SetupFrame { connection: [%s]; stream: [%s]; channel: [%s] }",
+                            connectionId, streamId, channel)));
               }
 
               CloseHelper.quietClose(serverManagementPublication);
@@ -170,8 +172,7 @@ public void onAvailableImage(Image image) {
 
                         final NanoClock nanoClock = aeron.context().nanoClock();
                         final long nowNs = nanoClock.nanoTime();
-                        final long deadlineNs =
-                            nowNs + aeron.context().keepAliveIntervalNs() * 1000;
+                        final long deadlineNs = nowNs + timeoutNs;
 
                         idleStrategy.reset();
                         for (; ; ) {
@@ -213,7 +214,12 @@ public void onAvailableImage(Image image) {
   public static AeronClientTransport createUdp(
       Aeron aeron, String host, int port, EventLoopGroup resources) {
     final Supplier<IdleStrategy> idleStrategySupplier =
-        () -> new BackoffIdleStrategy(/* maxSpins */ 100, /* maxYields */ 1000, /* minParkPeriodNs */ 10000, /* maxParkPeriodNs */100000);
+        () ->
+            new BackoffIdleStrategy(
+                /* maxSpins */ 100, /* maxYields */
+                1000, /* minParkPeriodNs */
+                10000, /* maxParkPeriodNs */
+                100000);
     return new AeronClientTransport(
         aeron,
         new ChannelUriStringBuilder()

rsocket-transport-aeron/src/main/java/io/rsocket/transport/aeron/WrappedDirectBufferByteBuf.java

@@ -230,10 +230,10 @@ public int nioBufferCount() {
 
   @Override
   public ByteBuffer nioBuffer(int index, int length) {
-    // FIXME: works incorrectly
     final ByteBuffer byteBuffer =
         BufferUtil.allocateDirectAligned(length, BitUtil.CACHE_LINE_LENGTH);
     directBuffer.getBytes(index, byteBuffer, length);
+    byteBuffer.flip();
     return byteBuffer;
   }
 

rsocket-transport-aeron/src/test/resources/logback-test.xml

@@ -25,7 +25,6 @@
 
     <logger name="io.rsocket.transport.netty" level="INFO"/>
     <logger name="io.rsocket.FrameLogger" level="INFO"/>
-    <logger name="io.rsocket.fragmentation.FragmentationDuplexConnection" level="INFO"/>
     <logger name="io.rsocket.transport.netty" level="INFO"/>
     <logger name="io.rsocket.core.RSocketRequester" level="DEBUG"/>
     <logger name="io.rsocket.core.RSocketResponder" level="DEBUG"/>