QUEUE-14 improved bytes queue ring buffer

OpenHFT · Feb 8, 2015 · 14e75cb · 14e75cb
1 parent a404283
commit 14e75cb
Showing 1 changed file with 27 additions and 19 deletions.
diff --git a/chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/ringbuffer/BytesQueue.java b/chronicle-queue/src/main/java/net/openhft/chronicle/queue/impl/ringbuffer/BytesQueue.java
@@ -50,11 +50,11 @@ public boolean offer(@NotNull Bytes bytes) throws InterruptedException {
                 queue.readupto.set(writeLocation);
                 queue.write(writeLocation, bytes.remaining());
 
-                long nextWriteLocation = queue.nextOffset(writeLocation, 8);
+                long offset = queue.nextOffset(writeLocation, 8);
 
-                nextWriteLocation = queue.write(bytes, nextWriteLocation);
+                offset = queue.write(bytes, offset);
 
-                writeLocation = nextWriteLocation;
+                writeLocation = offset;
                 return true;
 
             }
@@ -95,37 +95,26 @@ public Bytes poll(@NotNull Bytes using) throws InterruptedException, IllegalStat
                 continue;
 
             long elementSize = queue.readLong(offset);
-            if (elementSize == -1) {
-                using.position(0);
-                using.limit(0);
-                return using;
-            }
 
             // checks that the 'using' bytes is large enough
             checkSize(using, elementSize);
 
             offset = queue.nextOffset(offset, 8);
             assert offset < queue.capacity();
-            for (int i = 0; i < elementSize; offset = queue.blockForReadSpace(offset), i++) {
-
-                if (offset == -1) {
-                    using.position(0);
-                    using.limit(0);
-                    return null;
-                }
 
-                byte b = queue.read(offset);
+            using.limit(using.position() + elementSize);
-                using.write(b);
-            }
 
+            offset = queue.read(using, offset);
             queue.writeupto.set(offset);
             queue.readLocation.set(offset);
 
-            return using.flip();
+            return using;
         }
 
     }
 
+
+
     private static void checkSize(@NotNull Bytes using, long elementSize) {
         if (using.remaining() < elementSize)
             throw new IllegalStateException("requires size=" + elementSize + " " +
@@ -176,6 +165,25 @@ private long write(@NotNull Bytes bytes, long offset) {
 
         }
 
+        private long read(@NotNull Bytes bytes, long offset) {
+
+            long endOffSet = nextOffset(offset, bytes.remaining());
+
+            if (endOffSet >= offset) {
+                bytes.write(buffer, offset, bytes.remaining());
+                bytes.flip();
+                return endOffSet;
+            }
+
+            bytes.write(buffer, offset, capacity() - offset);
+            bytes.write(buffer, 1, bytes.remaining());
+            bytes.flip();
+
+            return endOffSet;
+
+        }
+
+
         boolean isBytesBigEndian() {
             try {
                 putLongB(0, 1);