Dispatcher extends Actor { protected Subscription doOpenSubscription( final String subscriptionName, final ActorCondition onConsumption) { ensureUniqueSubscriptionName(subscriptionName); LOG.trace("Open subscription with name '{}'", subscriptionName); final Subscription[] newSubscriptions = new Subscription[subscriptions.length + 1]; System.arraycopy(subscriptions, 0, newSubscriptions, 0, subscriptions.length); final int subscriberId = newSubscriptions.length - 1; final Subscription subscription = newSubscription(subscriberId, subscriptionName, onConsumption); newSubscriptions[subscriberId] = subscription; subscriptions = newSubscriptions; onConsumption.signal(); return subscription; } 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 shouldReadFragmentsFromPartition() { dispatcher.doOpenSubscription("test", mock(ActorCondition.class)); when(subscriberPosition.get()).thenReturn(0L); when(publisherPosition.get()).thenReturn(position(0, A_FRAGMENT_LENGTH)); doReturn(1) .when(subscriptionSpy) .pollFragments( logBufferPartition0, fragmentHandler, 0, 0, 2, position(0, A_FRAGMENT_LENGTH), false); final int fragmentsRead = subscriptionSpy.poll(fragmentHandler, 2); assertThat(fragmentsRead).isEqualTo(1); verify(subscriberPosition).get(); verify(subscriptionSpy) .pollFragments( logBufferPartition0, fragmentHandler, 0, 0, 2, position(0, A_FRAGMENT_LENGTH), false); }
@Test public void shouldNotReadBeyondPublisherPosition() { dispatcher.doOpenSubscription("test", mock(ActorCondition.class)); when(subscriptionSpy.getPosition()).thenReturn(0L); when(publisherPosition.get()).thenReturn(0L); final int fragmentsRead = subscriptionSpy.poll(fragmentHandler, 1); assertThat(fragmentsRead).isEqualTo(0); }
@Test public void shouldReadFragmentsFromPartitionOnPeekAndConsume() { dispatcher.doOpenSubscription("test", mock(ActorCondition.class)); when(subscriberPosition.get()).thenReturn(0L); when(publisherPosition.get()).thenReturn(position(0, A_FRAGMENT_LENGTH)); doReturn(1) .when(subscriptionSpy) .pollFragments( logBufferPartition0, fragmentHandler, 0, 0, 2, position(0, A_FRAGMENT_LENGTH), true); final int fragmentsRead = subscriptionSpy.peekAndConsume(fragmentHandler, 2); assertThat(fragmentsRead).isEqualTo(1); verify(subscriberPosition).get(); verify(subscriptionSpy) .pollFragments( logBufferPartition0, fragmentHandler, 0, 0, 2, position(0, A_FRAGMENT_LENGTH), true); }
@Test public void shouldNotOpenSubscriptionWithSameName() { dispatcher.doOpenSubscription("s1", mock(ActorCondition.class)); Assert.assertThrows( "subscription with name 's1' already exists", IllegalStateException.class, () -> dispatcher.doOpenSubscription("s1", mock(ActorCondition.class))); }