Dispatcher extends Actor { public long claimFragmentBatch( final ClaimedFragmentBatch batch, final int fragmentCount, final int batchLength) { return offer( (partition, activePartitionId) -> logAppender.claim( partition, activePartitionId, batch, fragmentCount, batchLength, onClaimComplete), fragmentCount, LogBufferAppender.claimedBatchLength(fragmentCount, batchLength)); } 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 shouldIncreasePositionByFragmentCountAfterClaimingBatch() { final int fragmentCount = 3; 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(claimedFragmentBatch), eq(fragmentCount), eq(A_MSG_PAYLOAD_LENGTH), any())) .thenReturn(A_FRAGMENT_LENGTH); long newPosition = dispatcher.claimFragmentBatch(claimedFragmentBatch, fragmentCount, A_MSG_PAYLOAD_LENGTH); assertThat(newPosition).isEqualTo(INITIAL_POSITION); newPosition = dispatcher.claimFragmentBatch(claimedFragmentBatch, fragmentCount, A_MSG_PAYLOAD_LENGTH); assertThat(newPosition).isEqualTo(INITIAL_POSITION + fragmentCount); }