@Test @DisplayName("single shard and return stats when picked up the shard " + "and `Optional.empty()` if shard was already picked") public void returnOptionalEmptyIfPicked() { int shardCount = 11; ShardedWorkRegistry registry = new InMemoryShardedWorkRegistry(); FixedShardStrategy strategy = new FixedShardStrategy(shardCount); Delivery delivery = Delivery.newBuilder() .setStrategy(strategy) .setWorkRegistry(registry) .build(); ServerEnvironment env = ServerEnvironment.instance(); env.use(delivery, Tests.class); ShardIndex index = strategy.nonEmptyShard(); TenantId tenantId = GivenTenantId.generate(); TenantAwareRunner.with(tenantId) .run(() -> assertStatsMatch(delivery, index)); Optional<ShardProcessingSession> session = registry.pickUp(index, env.nodeId()); assertThat(session).isPresent(); TenantAwareRunner.with(tenantId) .run(() -> assertStatsEmpty(delivery, index)); }
public static DeliveryBuilder newBuilder() { return new DeliveryBuilder(); }
Delivery implements Logging { public static DeliveryBuilder newBuilder() { return new DeliveryBuilder(); } }
Delivery implements Logging { public static DeliveryBuilder newBuilder() { return new DeliveryBuilder(); } Delivery(DeliveryBuilder builder); }
Delivery implements Logging { public static DeliveryBuilder newBuilder() { return new DeliveryBuilder(); } Delivery(DeliveryBuilder builder); static DeliveryBuilder newBuilder(); static Delivery local(); static Delivery localAsync(); Optional<DeliveryStats> deliverMessagesFrom(ShardIndex index); Inbox.Builder<I> newInbox(TypeUrl entityType); CatchUpProcessBuilder<I> newCatchUpProcess(ProjectionRepository<I, ?, ?> repo); @Internal void registerDispatchersIn(BoundedContext context); @Internal MulticastDispatchListener dispatchListener(); void subscribe(ShardObserver observer); }
Delivery implements Logging { public static DeliveryBuilder newBuilder() { return new DeliveryBuilder(); } Delivery(DeliveryBuilder builder); static DeliveryBuilder newBuilder(); static Delivery local(); static Delivery localAsync(); Optional<DeliveryStats> deliverMessagesFrom(ShardIndex index); Inbox.Builder<I> newInbox(TypeUrl entityType); CatchUpProcessBuilder<I> newCatchUpProcess(ProjectionRepository<I, ?, ?> repo); @Internal void registerDispatchersIn(BoundedContext context); @Internal MulticastDispatchListener dispatchListener(); void subscribe(ShardObserver observer);  }