Configure same byte order for intermediate buffer in StreamDefragger.

js-labs · js-labs · commit 683cbbe5908d · 2022-02-17T22:02:44.000+03:00
diff --git a/src/main/java/org/jsl/collider/StreamDefragger.java b/src/main/java/org/jsl/collider/StreamDefragger.java
@@ -80,29 +80,28 @@ public abstract class StreamDefragger
 
     private static class BufferImpl extends RetainableByteBufferImpl
     {
-        public BufferImpl( ByteBuffer byteBuffer )
+        public BufferImpl(ByteBuffer byteBuffer)
         {
-            super( byteBuffer );
+            super(byteBuffer);
         }
 
         protected void finalRelease()
         {
             /* Do nothing */
         }
 
-        public static BufferImpl create( int capacity, boolean isDirect )
+        public static BufferImpl create(int capacity, boolean isDirect)
         {
             if (capacity < 1024)
                 capacity = 1024;
             final ByteBuffer byteBuffer =
-                    isDirect ? ByteBuffer.allocateDirect( capacity )
-                             : ByteBuffer.allocate( capacity );
-            return new BufferImpl( byteBuffer );
+                    isDirect ? ByteBuffer.allocateDirect(capacity)
+                             : ByteBuffer.allocate(capacity);
+            return new BufferImpl(byteBuffer);
         }
     }
 
-    private static boolean copyData(
-            RetainableByteBuffer rdst, RetainableByteBuffer rsrc, int bytes )
+    private static boolean copyData(RetainableByteBuffer rdst, RetainableByteBuffer rsrc, int bytes)
     {
         final ByteBuffer dst = rdst.getNioByteBuffer();
         final ByteBuffer src = rsrc.getNioByteBuffer();
@@ -111,26 +110,26 @@ private static boolean copyData(
         final int available = (limit - pos);
         if (available < bytes)
         {
-            dst.put( src );
+            dst.put(src);
             return false;
         }
         else
         {
-            src.limit( pos + bytes );
-            dst.put( src );
-            src.limit( limit );
+            src.limit(pos + bytes);
+            dst.put(src);
+            src.limit(limit);
             return true;
         }
     }
 
-    public StreamDefragger( int headerSize )
+    public StreamDefragger(int headerSize)
     {
         m_headerSize = headerSize;
     }
 
-    public final RetainableByteBuffer getNext( RetainableByteBuffer data )
+    public final RetainableByteBuffer getNext(RetainableByteBuffer data)
     {
-        assert( m_data == null );
+        assert(m_data == null);
 
         if ((m_buf != null) && (m_buf.position() > 0))
         {
@@ -142,24 +141,26 @@ public final RetainableByteBuffer getNext( RetainableByteBuffer data )
                     return null;
 
                 m_buf.flip();
-                m_packetLen = validateHeader( m_buf.getNioByteBuffer() );
+                m_packetLen = validateHeader(m_buf.getNioByteBuffer());
                 if (m_packetLen < 0)
                     return INVALID_HEADER;
                 assert(m_packetLen >= m_headerSize);
 
                 if (m_buf.capacity() < m_packetLen)
                 {
-                    final BufferImpl buf = BufferImpl.create( m_packetLen, data.getNioByteBuffer().isDirect() );
-                    m_buf.position( 0 );
-                    m_buf.limit( m_headerSize );
-                    buf.put( m_buf );
+                    final ByteBuffer dataByteBuffer = data.getNioByteBuffer();
+                    final BufferImpl buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());
+                    buf.order(dataByteBuffer.order());
+                    m_buf.position(0);
+                    m_buf.limit(m_headerSize);
+                    buf.put(m_buf);
                     m_buf = buf;
                 }
                 else
-                    m_buf.limit( m_buf.capacity() );
+                    m_buf.limit(m_buf.capacity());
 
                 pos = m_headerSize;
-                m_buf.position( pos );
+                m_buf.position(pos);
             }
 
             final int cc = (m_packetLen - pos);
@@ -182,8 +183,8 @@ public final RetainableByteBuffer getNext( RetainableByteBuffer data )
 
     public final RetainableByteBuffer getNext()
     {
-        m_data.position( m_pos );
-        m_data.limit( m_limit );
+        m_data.position(m_pos);
+        m_data.limit(m_limit);
 
         final int bytesRemaining = (m_limit - m_pos);
         if (bytesRemaining == 0)
@@ -194,41 +195,46 @@ public final RetainableByteBuffer getNext()
             return null;
         }
 
+        final ByteBuffer dataByteBuffer = m_data.getNioByteBuffer();
         if (bytesRemaining < m_headerSize)
         {
             /* m_buf will always have at least m_headerSize capacity */
             if ((m_buf == null) || !m_buf.releaseReuse())
-                m_buf = BufferImpl.create( m_headerSize, m_data.getNioByteBuffer().isDirect() );
-            m_buf.put( m_data );
+            {
+                m_buf = BufferImpl.create(m_headerSize, dataByteBuffer.isDirect());
+                m_buf.order(dataByteBuffer.order());
+            }
+            m_buf.put(m_data);
             m_data = null;
             return null;
         }
 
-        m_packetLen = validateHeader( m_data.getNioByteBuffer() );
-        m_data.position( m_pos );
+        m_packetLen = validateHeader(dataByteBuffer);
+        m_data.position(m_pos);
 
         if (m_packetLen <= 0)
             return INVALID_HEADER;
 
         if (bytesRemaining < m_packetLen)
         {
             if (m_buf == null)
-                m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );
+                m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());
             else if (m_buf.capacity() < m_packetLen)
             {
                 m_buf.release();
-                m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );
+                m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());
             }
             else if (!m_buf.releaseReuse())
-                m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );
+                m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());
 
-            m_buf.put( m_data );
+            m_buf.order(dataByteBuffer.order());
+            m_buf.put(m_data);
             m_data = null;
             return null;
         }
 
         m_pos += m_packetLen;
-        m_data.limit( m_pos );
+        m_data.limit(m_pos);
         return m_data;
     }
 
@@ -241,5 +247,5 @@ public void close()
         }
     }
 
-    abstract protected int validateHeader( ByteBuffer header );
+    abstract protected int validateHeader(ByteBuffer header);
 }
diff --git a/tests/src/org/jsl/tests/echo_latency/Server.java b/tests/src/org/jsl/tests/echo_latency/Server.java
@@ -103,7 +103,7 @@ private class TestAcceptor extends Acceptor
     {
         public TestAcceptor()
         {
-            super( new InetSocketAddress(0) );
+            super(0);
             //inputQueueBlockSize = 700;
         }
 
diff --git a/tests/src/org/jsl/tests/echo_throughput/Server.java b/tests/src/org/jsl/tests/echo_throughput/Server.java
@@ -1,6 +1,6 @@
 /*
  * JS-Collider framework tests.
- * Copyright (C) 2013 Sergey Zubarev
+ * Copyright (C) 2022 Sergey Zubarev
  * info@js-labs.org
  *
  * This program is free software: you can redistribute it and/or modify
@@ -133,7 +133,7 @@ private class TestAcceptor extends Acceptor
     {
         public TestAcceptor( int socketBufferSize )
         {
-            super( new InetSocketAddress(0) );
+            super(0);
             socketRecvBufSize = socketBufferSize;
             socketSendBufSize = socketBufferSize;
         }
diff --git a/tests/src/org/jsl/tests/msg_size_eq_block_size/Server.java b/tests/src/org/jsl/tests/msg_size_eq_block_size/Server.java
@@ -1,6 +1,6 @@
 /*
  * JS-Collider framework tests.
- * Copyright (C) 2018 Sergey Zubarev
+ * Copyright (C) 2022 Sergey Zubarev
  * info@js-labs.org
  *
  * This program is free software: you can redistribute it and/or modify
@@ -87,7 +87,7 @@ private class TestAcceptor extends Acceptor
     {
         TestAcceptor()
         {
-            super(new InetSocketAddress(0));
+            super(0);
         }
 
         public void onAcceptorStarted(Collider collider, int localPort)
diff --git a/tests/src/org/jsl/tests/send_throughput/Server.java b/tests/src/org/jsl/tests/send_throughput/Server.java
@@ -1,6 +1,6 @@
 /*
  * JS-Collider framework tests.
- * Copyright (C) 2013 Sergey Zubarev
+ * Copyright (C) 2022 Sergey Zubarev
  * info@js-labs.org
  *
  * This program is free software: you can redistribute it and/or modify
@@ -132,7 +132,7 @@ private class TestAcceptor extends Acceptor
 
         public TestAcceptor(int port, int socketBufferSize, Client client)
         {
-            super(new InetSocketAddress(port));
+            super(port);
             this.tcpNoDelay = true;
             this.socketSendBufSize = socketBufferSize;
             this.m_client = client;
diff --git a/tests/src/org/jsl/tests/session_latency/Server.java b/tests/src/org/jsl/tests/session_latency/Server.java
@@ -1,6 +1,6 @@
 /*
  * JS-Collider framework tests.
- * Copyright (C) 2013 Sergey Zubarev
+ * Copyright (C) 2022 Sergey Zubarev
  * info@js-labs.org
  *
  * This program is free software: you can redistribute it and/or modify
@@ -114,7 +114,7 @@ private class TestAcceptor extends Acceptor
     {
         public TestAcceptor()
         {
-            super( new InetSocketAddress(0) );
+            super(0);
             tcpNoDelay = true;
         }
 
diff --git a/tests/src/org/jsl/tests/session_throughput/Server.java b/tests/src/org/jsl/tests/session_throughput/Server.java
@@ -1,6 +1,6 @@
 /*
  * JS-Collider framework tests.
- * Copyright (C) 2013 Sergey Zubarev
+ * Copyright (C) 2022 Sergey Zubarev
  * info@js-labs.org
  *
  * This program is free software: you can redistribute it and/or modify
@@ -128,7 +128,7 @@ private class TestAcceptor extends Acceptor
     {
         public TestAcceptor()
         {
-            super( new InetSocketAddress(0) );
+            super(0);
             tcpNoDelay = true;
             socketRecvBufSize = m_socketBufferSize;
             socketSendBufSize = m_socketBufferSize;

Original file line number	Diff line number	Diff line change
`@@ -80,29 +80,28 @@ public abstract class StreamDefragger`
`80`	`80`
`81`	`81`	`private static class BufferImpl extends RetainableByteBufferImpl`
`82`	`82`	`{`
`83`		`- public BufferImpl( ByteBuffer byteBuffer )`
	`83`	`+ public BufferImpl(ByteBuffer byteBuffer)`
`84`	`84`	`{`
`85`		`- super( byteBuffer );`
	`85`	`+ super(byteBuffer);`
`86`	`86`	`}`
`87`	`87`
`88`	`88`	`protected void finalRelease()`
`89`	`89`	`{`
`90`	`90`	`/* Do nothing */`
`91`	`91`	`}`
`92`	`92`
`93`		`- public static BufferImpl create( int capacity, boolean isDirect )`
	`93`	`+ public static BufferImpl create(int capacity, boolean isDirect)`
`94`	`94`	`{`
`95`	`95`	`if (capacity < 1024)`
`96`	`96`	`capacity = 1024;`
`97`	`97`	`final ByteBuffer byteBuffer =`
`98`		`- isDirect ? ByteBuffer.allocateDirect( capacity )`
`99`		`- : ByteBuffer.allocate( capacity );`
`100`		`- return new BufferImpl( byteBuffer );`
	`98`	`+ isDirect ? ByteBuffer.allocateDirect(capacity)`
	`99`	`+ : ByteBuffer.allocate(capacity);`
	`100`	`+ return new BufferImpl(byteBuffer);`
`101`	`101`	`}`
`102`	`102`	`}`
`103`	`103`
`104`		`- private static boolean copyData(`
`105`		`- RetainableByteBuffer rdst, RetainableByteBuffer rsrc, int bytes )`
	`104`	`+ private static boolean copyData(RetainableByteBuffer rdst, RetainableByteBuffer rsrc, int bytes)`
`106`	`105`	`{`
`107`	`106`	`final ByteBuffer dst = rdst.getNioByteBuffer();`
`108`	`107`	`final ByteBuffer src = rsrc.getNioByteBuffer();`
`@@ -111,26 +110,26 @@ private static boolean copyData(`
`111`	`110`	`final int available = (limit - pos);`
`112`	`111`	`if (available < bytes)`
`113`	`112`	`{`
`114`		`- dst.put( src );`
	`113`	`+ dst.put(src);`
`115`	`114`	`return false;`
`116`	`115`	`}`
`117`	`116`	`else`
`118`	`117`	`{`
`119`		`- src.limit( pos + bytes );`
`120`		`- dst.put( src );`
`121`		`- src.limit( limit );`
	`118`	`+ src.limit(pos + bytes);`
	`119`	`+ dst.put(src);`
	`120`	`+ src.limit(limit);`
`122`	`121`	`return true;`
`123`	`122`	`}`
`124`	`123`	`}`
`125`	`124`
`126`		`- public StreamDefragger( int headerSize )`
	`125`	`+ public StreamDefragger(int headerSize)`
`127`	`126`	`{`
`128`	`127`	`m_headerSize = headerSize;`
`129`	`128`	`}`
`130`	`129`
`131`		`- public final RetainableByteBuffer getNext( RetainableByteBuffer data )`
	`130`	`+ public final RetainableByteBuffer getNext(RetainableByteBuffer data)`
`132`	`131`	`{`
`133`		`- assert( m_data == null );`
	`132`	`+ assert(m_data == null);`
`134`	`133`
`135`	`134`	`if ((m_buf != null) && (m_buf.position() > 0))`
`136`	`135`	`{`
`@@ -142,24 +141,26 @@ public final RetainableByteBuffer getNext( RetainableByteBuffer data )`
`142`	`141`	`return null;`
`143`	`142`
`144`	`143`	`m_buf.flip();`
`145`		`- m_packetLen = validateHeader( m_buf.getNioByteBuffer() );`
	`144`	`+ m_packetLen = validateHeader(m_buf.getNioByteBuffer());`
`146`	`145`	`if (m_packetLen < 0)`
`147`	`146`	`return INVALID_HEADER;`
`148`	`147`	`assert(m_packetLen >= m_headerSize);`
`149`	`148`
`150`	`149`	`if (m_buf.capacity() < m_packetLen)`
`151`	`150`	`{`
`152`		`- final BufferImpl buf = BufferImpl.create( m_packetLen, data.getNioByteBuffer().isDirect() );`
`153`		`- m_buf.position( 0 );`
`154`		`- m_buf.limit( m_headerSize );`
`155`		`- buf.put( m_buf );`
	`151`	`+ final ByteBuffer dataByteBuffer = data.getNioByteBuffer();`
	`152`	`+ final BufferImpl buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());`
	`153`	`+ buf.order(dataByteBuffer.order());`
	`154`	`+ m_buf.position(0);`
	`155`	`+ m_buf.limit(m_headerSize);`
	`156`	`+ buf.put(m_buf);`
`156`	`157`	`m_buf = buf;`
`157`	`158`	`}`
`158`	`159`	`else`
`159`		`- m_buf.limit( m_buf.capacity() );`
	`160`	`+ m_buf.limit(m_buf.capacity());`
`160`	`161`
`161`	`162`	`pos = m_headerSize;`
`162`		`- m_buf.position( pos );`
	`163`	`+ m_buf.position(pos);`
`163`	`164`	`}`
`164`	`165`
`165`	`166`	`final int cc = (m_packetLen - pos);`
`@@ -182,8 +183,8 @@ public final RetainableByteBuffer getNext( RetainableByteBuffer data )`
`182`	`183`
`183`	`184`	`public final RetainableByteBuffer getNext()`
`184`	`185`	`{`
`185`		`- m_data.position( m_pos );`
`186`		`- m_data.limit( m_limit );`
	`186`	`+ m_data.position(m_pos);`
	`187`	`+ m_data.limit(m_limit);`
`187`	`188`
`188`	`189`	`final int bytesRemaining = (m_limit - m_pos);`
`189`	`190`	`if (bytesRemaining == 0)`
`@@ -194,41 +195,46 @@ public final RetainableByteBuffer getNext()`
`194`	`195`	`return null;`
`195`	`196`	`}`
`196`	`197`
	`198`	`+ final ByteBuffer dataByteBuffer = m_data.getNioByteBuffer();`
`197`	`199`	`if (bytesRemaining < m_headerSize)`
`198`	`200`	`{`
`199`	`201`	`/* m_buf will always have at least m_headerSize capacity */`
`200`	`202`	`if ((m_buf == null) \|\| !m_buf.releaseReuse())`
`201`		`- m_buf = BufferImpl.create( m_headerSize, m_data.getNioByteBuffer().isDirect() );`
`202`		`- m_buf.put( m_data );`
	`203`	`+ {`
	`204`	`+ m_buf = BufferImpl.create(m_headerSize, dataByteBuffer.isDirect());`
	`205`	`+ m_buf.order(dataByteBuffer.order());`
	`206`	`+ }`
	`207`	`+ m_buf.put(m_data);`
`203`	`208`	`m_data = null;`
`204`	`209`	`return null;`
`205`	`210`	`}`
`206`	`211`
`207`		`- m_packetLen = validateHeader( m_data.getNioByteBuffer() );`
`208`		`- m_data.position( m_pos );`
	`212`	`+ m_packetLen = validateHeader(dataByteBuffer);`
	`213`	`+ m_data.position(m_pos);`
`209`	`214`
`210`	`215`	`if (m_packetLen <= 0)`
`211`	`216`	`return INVALID_HEADER;`
`212`	`217`
`213`	`218`	`if (bytesRemaining < m_packetLen)`
`214`	`219`	`{`
`215`	`220`	`if (m_buf == null)`
`216`		`- m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );`
	`221`	`+ m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());`
`217`	`222`	`else if (m_buf.capacity() < m_packetLen)`
`218`	`223`	`{`
`219`	`224`	`m_buf.release();`
`220`		`- m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );`
	`225`	`+ m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());`
`221`	`226`	`}`
`222`	`227`	`else if (!m_buf.releaseReuse())`
`223`		`- m_buf = BufferImpl.create( m_packetLen, m_data.getNioByteBuffer().isDirect() );`
	`228`	`+ m_buf = BufferImpl.create(m_packetLen, dataByteBuffer.isDirect());`
`224`	`229`
`225`		`- m_buf.put( m_data );`
	`230`	`+ m_buf.order(dataByteBuffer.order());`
	`231`	`+ m_buf.put(m_data);`
`226`	`232`	`m_data = null;`
`227`	`233`	`return null;`
`228`	`234`	`}`
`229`	`235`
`230`	`236`	`m_pos += m_packetLen;`
`231`		`- m_data.limit( m_pos );`
	`237`	`+ m_data.limit(m_pos);`
`232`	`238`	`return m_data;`
`233`	`239`	`}`
`234`	`240`
`@@ -241,5 +247,5 @@ public void close()`
`241`	`247`	`}`
`242`	`248`	`}`
`243`	`249`
`244`		`- abstract protected int validateHeader( ByteBuffer header );`
	`250`	`+ abstract protected int validateHeader(ByteBuffer header);`
`245`	`251`	`}`
Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,7 @@ private class TestAcceptor extends Acceptor`
`103`	`103`	`{`
`104`	`104`	`public TestAcceptor()`
`105`	`105`	`{`
`106`		`- super( new InetSocketAddress(0) );`
	`106`	`+ super(0);`
`107`	`107`	`//inputQueueBlockSize = 700;`
`108`	`108`	`}`
`109`	`109`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`/*`
`2`	`2`	`* JS-Collider framework tests.`
`3`		`- * Copyright (C) 2013 Sergey Zubarev`
	`3`	`+ * Copyright (C) 2022 Sergey Zubarev`
`4`	`4`	`* info@js-labs.org`
`5`	`5`	`*`
`6`	`6`	`* This program is free software: you can redistribute it and/or modify`
`@@ -133,7 +133,7 @@ private class TestAcceptor extends Acceptor`
`133`	`133`	`{`
`134`	`134`	`public TestAcceptor( int socketBufferSize )`
`135`	`135`	`{`
`136`		`- super( new InetSocketAddress(0) );`
	`136`	`+ super(0);`
`137`	`137`	`socketRecvBufSize = socketBufferSize;`
`138`	`138`	`socketSendBufSize = socketBufferSize;`
`139`	`139`	`}`