@Test public void shouldUnblockWhenFullWithoutHeader() { final int messageLength = ALIGNMENT * 4; when(buffer.getLongVolatile(HEAD_COUNTER_INDEX)).thenReturn((long)messageLength); when(buffer.getLongVolatile(TAIL_COUNTER_INDEX)).thenReturn((long)messageLength + CAPACITY); when(buffer.getIntVolatile(messageLength * 2)).thenReturn(messageLength); assertTrue(ringBuffer.unblock()); final InOrder inOrder = inOrder(buffer); inOrder.verify(buffer).putInt(typeOffset(messageLength), PADDING_MSG_TYPE_ID); inOrder.verify(buffer).putIntOrdered(lengthOffset(messageLength), messageLength); }
public boolean unblock() { final AtomicBuffer buffer = this.buffer; final long headPosition = buffer.getLongVolatile(headPositionIndex); final long tailPosition = buffer.getLongVolatile(tailPositionIndex); if (headPosition == tailPosition) { return false; } final int mask = capacity - 1; final int consumerIndex = (int)(headPosition & mask); final int producerIndex = (int)(tailPosition & mask); boolean unblocked = false; int length = buffer.getIntVolatile(consumerIndex); if (length < 0) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), -length); unblocked = true; } else if (0 == length) { final int limit = producerIndex > consumerIndex ? producerIndex : capacity; int i = consumerIndex + ALIGNMENT; do { length = buffer.getIntVolatile(i); if (0 != length) { if (scanBackToConfirmStillZeroed(buffer, i, consumerIndex)) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), i - consumerIndex); unblocked = true; } break; } i += ALIGNMENT; } while (i < limit); } return unblocked; }
ManyToOneRingBuffer implements RingBuffer { public boolean unblock() { final AtomicBuffer buffer = this.buffer; final long headPosition = buffer.getLongVolatile(headPositionIndex); final long tailPosition = buffer.getLongVolatile(tailPositionIndex); if (headPosition == tailPosition) { return false; } final int mask = capacity - 1; final int consumerIndex = (int)(headPosition & mask); final int producerIndex = (int)(tailPosition & mask); boolean unblocked = false; int length = buffer.getIntVolatile(consumerIndex); if (length < 0) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), -length); unblocked = true; } else if (0 == length) { final int limit = producerIndex > consumerIndex ? producerIndex : capacity; int i = consumerIndex + ALIGNMENT; do { length = buffer.getIntVolatile(i); if (0 != length) { if (scanBackToConfirmStillZeroed(buffer, i, consumerIndex)) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), i - consumerIndex); unblocked = true; } break; } i += ALIGNMENT; } while (i < limit); } return unblocked; } }
ManyToOneRingBuffer implements RingBuffer { public boolean unblock() { final AtomicBuffer buffer = this.buffer; final long headPosition = buffer.getLongVolatile(headPositionIndex); final long tailPosition = buffer.getLongVolatile(tailPositionIndex); if (headPosition == tailPosition) { return false; } final int mask = capacity - 1; final int consumerIndex = (int)(headPosition & mask); final int producerIndex = (int)(tailPosition & mask); boolean unblocked = false; int length = buffer.getIntVolatile(consumerIndex); if (length < 0) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), -length); unblocked = true; } else if (0 == length) { final int limit = producerIndex > consumerIndex ? producerIndex : capacity; int i = consumerIndex + ALIGNMENT; do { length = buffer.getIntVolatile(i); if (0 != length) { if (scanBackToConfirmStillZeroed(buffer, i, consumerIndex)) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), i - consumerIndex); unblocked = true; } break; } i += ALIGNMENT; } while (i < limit); } return unblocked; } ManyToOneRingBuffer(final AtomicBuffer buffer); }
ManyToOneRingBuffer implements RingBuffer { public boolean unblock() { final AtomicBuffer buffer = this.buffer; final long headPosition = buffer.getLongVolatile(headPositionIndex); final long tailPosition = buffer.getLongVolatile(tailPositionIndex); if (headPosition == tailPosition) { return false; } final int mask = capacity - 1; final int consumerIndex = (int)(headPosition & mask); final int producerIndex = (int)(tailPosition & mask); boolean unblocked = false; int length = buffer.getIntVolatile(consumerIndex); if (length < 0) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), -length); unblocked = true; } else if (0 == length) { final int limit = producerIndex > consumerIndex ? producerIndex : capacity; int i = consumerIndex + ALIGNMENT; do { length = buffer.getIntVolatile(i); if (0 != length) { if (scanBackToConfirmStillZeroed(buffer, i, consumerIndex)) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), i - consumerIndex); unblocked = true; } break; } i += ALIGNMENT; } while (i < limit); } return unblocked; } ManyToOneRingBuffer(final AtomicBuffer buffer); int capacity(); boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, final int length); int read(final MessageHandler handler); int read(final MessageHandler handler, final int messageCountLimit); int maxMsgLength(); long nextCorrelationId(); AtomicBuffer buffer(); void consumerHeartbeatTime(final long time); long consumerHeartbeatTime(); long producerPosition(); long consumerPosition(); int size(); boolean unblock(); int tryClaim(final int msgTypeId, final int length); void commit(final int index); void abort(final int index); }
ManyToOneRingBuffer implements RingBuffer { public boolean unblock() { final AtomicBuffer buffer = this.buffer; final long headPosition = buffer.getLongVolatile(headPositionIndex); final long tailPosition = buffer.getLongVolatile(tailPositionIndex); if (headPosition == tailPosition) { return false; } final int mask = capacity - 1; final int consumerIndex = (int)(headPosition & mask); final int producerIndex = (int)(tailPosition & mask); boolean unblocked = false; int length = buffer.getIntVolatile(consumerIndex); if (length < 0) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), -length); unblocked = true; } else if (0 == length) { final int limit = producerIndex > consumerIndex ? producerIndex : capacity; int i = consumerIndex + ALIGNMENT; do { length = buffer.getIntVolatile(i); if (0 != length) { if (scanBackToConfirmStillZeroed(buffer, i, consumerIndex)) { buffer.putInt(typeOffset(consumerIndex), PADDING_MSG_TYPE_ID); buffer.putIntOrdered(lengthOffset(consumerIndex), i - consumerIndex); unblocked = true; } break; } i += ALIGNMENT; } while (i < limit); } return unblocked; } ManyToOneRingBuffer(final AtomicBuffer buffer); int capacity(); boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final int offset, final int length); int read(final MessageHandler handler); int read(final MessageHandler handler, final int messageCountLimit); int maxMsgLength(); long nextCorrelationId(); AtomicBuffer buffer(); void consumerHeartbeatTime(final long time); long consumerHeartbeatTime(); long producerPosition(); long consumerPosition(); int size(); boolean unblock(); int tryClaim(final int msgTypeId, final int length); void commit(final int index); void abort(final int index);  }