Dispatcher extends Actor { public long claimSingleFragment(final ClaimedFragment claim, final int length) { return claimSingleFragment(claim, length, 0); } Dispatcher(
      final LogBuffer logBuffer,
      final LogBufferAppender logAppender,
      final AtomicPosition publisherLimit,
      final AtomicPosition publisherPosition,
      final long initialPosition,
      final int logWindowLength,
      final int maxFragmentLength,
      final String[] subscriptionNames,
      final String name); @Override String getName(); long claimSingleFragment(final ClaimedFragment claim, final int length); long claimSingleFragment(
      final ClaimedFragment claim, final int length, final int streamId); long claimFragmentBatch(
      final ClaimedFragmentBatch batch, final int fragmentCount, final int batchLength); int updatePublisherLimit(); Subscription openSubscription(final String subscriptionName); ActorFuture<Subscription> openSubscriptionAsync(final String subscriptionName); boolean isClosed(); LogBuffer getLogBuffer(); int getMaxFragmentLength(); long getPublisherPosition(); @Override String toString();  }
@Test public void shouldNotClaimBeyondPublisherLimit() { when(logBuffer.getActivePartitionIdVolatile()).thenReturn(0); when(logBufferPartition0.getTailCounterVolatile()).thenReturn(0); when(publisherLimit.get()).thenReturn(position(0, 0)); final long newPosition = dispatcher.claimSingleFragment(claimedFragment, A_MSG_PAYLOAD_LENGTH); assertThat(newPosition).isEqualTo(-1); verify(publisherLimit).get(); verifyNoMoreInteractions(publisherLimit); verifyNoMoreInteractions(logAppender); verifyNoMoreInteractions(claimedFragment); verify(logBuffer).getActivePartitionIdVolatile(); verify(logBuffer).getPartition(0); verify(logBufferPartition0).getTailCounterVolatile(); }
@Test public void shouldClaimFragment() { when(logBuffer.getActivePartitionIdVolatile()).thenReturn(0); when(logBufferPartition0.getTailCounterVolatile()).thenReturn(0); when(publisherLimit.get()).thenReturn(position(0, A_FRAGMENT_LENGTH)); when(logAppender.claim( eq(logBufferPartition0), eq(0), eq(claimedFragment), eq(A_MSG_PAYLOAD_LENGTH), eq(A_STREAM_ID), any())) .thenReturn(A_FRAGMENT_LENGTH); final long newPosition = dispatcher.claimSingleFragment(claimedFragment, A_MSG_PAYLOAD_LENGTH, A_STREAM_ID); assertThat(newPosition).isEqualTo(INITIAL_POSITION); verify(logAppender) .claim( eq(logBufferPartition0), eq(0), eq(claimedFragment), eq(A_MSG_PAYLOAD_LENGTH), eq(A_STREAM_ID), Mockito.any()); verify(publisherLimit).get(); verify(publisherPosition).proposeMaxOrdered(position(0, A_FRAGMENT_LENGTH)); verify(logBuffer).getActivePartitionIdVolatile(); verify(logBuffer).getPartition(0); verify(logBufferPartition0).getTailCounterVolatile(); }
@Test public void shouldIncrementRecordPositionAfterClaimingFragment() { when(publisherLimit.get()).thenReturn(position(0, A_FRAGMENT_LENGTH)); when(logAppender.claim( eq(logBufferPartition0), eq(0), eq(claimedFragment), eq(A_MSG_PAYLOAD_LENGTH), eq(A_STREAM_ID), any())) .thenReturn(A_FRAGMENT_LENGTH); long newPosition = dispatcher.claimSingleFragment(claimedFragment, A_MSG_PAYLOAD_LENGTH, A_STREAM_ID); assertThat(newPosition).isEqualTo(INITIAL_POSITION); newPosition = dispatcher.claimSingleFragment(claimedFragment, A_MSG_PAYLOAD_LENGTH, A_STREAM_ID); assertThat(newPosition).isEqualTo(INITIAL_POSITION + 1); }