@Test @DisplayName("a single shard to a single target in a multi-threaded env in batches") public void deliverInBatch() { FixedShardStrategy strategy = new FixedShardStrategy(1); MemoizingDeliveryMonitor monitor = new MemoizingDeliveryMonitor(); int pageSize = 20; Delivery delivery = Delivery.newBuilder() .setStrategy(strategy) .setDeduplicationWindow(Durations.ZERO) .setMonitor(monitor) .setPageSize(pageSize) .build(); deliverAfterPause(delivery); ServerEnvironment.instance() .use(delivery, Tests.class); ImmutableSet<String> targets = singleTarget(); NastyClient simulator = new NastyClient(7, false); simulator.runWith(targets); String theTarget = targets.iterator() .next(); int signalsDispatched = simulator.signalsPerTarget() .get(theTarget) .size(); assertThat(simulator.callsToRepoLoadOrCreate(theTarget)).isLessThan(signalsDispatched); assertThat(simulator.callsToRepoStore(theTarget)).isLessThan(signalsDispatched); assertStages(monitor, pageSize); }
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);  }