@Test void testMultipleUpdates() { TopologyContext<String, TestCommand, TestEvent, Optional<TestAggregate>> ctx = ctxBuilder.buildContext(); driver = new TestDriverInitializer().build(builder -> EventSourcedTopology.addTopology(ctx, builder)); TestContextDriver<String, TestCommand, TestEvent, Optional<TestAggregate>> ctxDriver = new TestContextDriver<>(ctx, driver); ctxDriver.publishCommand( key, CommandRequest.of( CommandId.random(), key, Sequence.first(), new TestCommand.CreateCommand("firstName"))); ctxDriver.verifyAggregateUpdate(key, null); CommandResponse<String> response = ctxDriver.verifyCommandResponse(key, null); ctxDriver.verifyEvents(key, null); for (int i = 0; i < 10; i++) { String newName = String.format("firstName %d", i); Sequence lastSequence = response.sequenceResult().getOrElse(Sequence.first()); CommandRequest<String, TestCommand> commandRequest = CommandRequest.of( CommandId.random(), key, lastSequence, new TestCommand.UpdateWithNothingCommand(newName)); ctxDriver.publishCommand(key, commandRequest); List<ValueWithSequence<TestEvent>> events = ctxDriver.verifyEvents(key, iV -> { Integer index = iV.v1(); ValueWithSequence<TestEvent> value = iV.v2(); assertThat(value.sequence().getSeq()).isEqualTo(index.longValue() + lastSequence.getSeq() + 1L); }); assertThat(events.size()).isGreaterThan(1); long ms = 0L; for (ValueWithSequence<TestEvent> event: events) { long es = event.sequence().getSeq(); if (ms < es) ms = es; } final long maxSequence = ms; response = ctxDriver.verifyCommandResponse(key, v -> { assertThat(v.sequenceResult().isSuccess()).isEqualTo(true); assertThat(v.sequenceResult().getOrElse(Sequence.first()).getSeq()).isEqualTo(maxSequence); }); ctxDriver.verifyAggregateUpdate(key, update -> { assertThat(update.sequence().getSeq()).isEqualTo(maxSequence); assertThat(update.aggregate().isPresent()).isEqualTo(true); assertThat(update.aggregate().get().name()).isEqualTo(newName); }); ctxDriver.verifyNoCommandResponse(); ctxDriver.verifyNoAggregateUpdate(); } }
public static <K, C, E, A> InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder) { final KStream<K, CommandRequest<K, C>> commandRequestStream = EventSourcedConsumer.commandRequestStream(ctx, builder); final KStream<K, CommandResponse<K>> commandResponseStream = EventSourcedConsumer.commandResponseStream(ctx, builder); final KTable<K, AggregateUpdate<A>> aggregateTable = EventSourcedConsumer.aggregateTable(ctx, builder); final DistributorContext<CommandId, CommandResponse<K>> distCtx = getDistributorContext(ctx); final KStream<CommandId, String> resultsTopicMapStream = ResultDistributor.resultTopicMapStream(distCtx, builder); Tuple2<KStream<K, CommandRequest<K, C>>, KStream<K, CommandResponse<K>>> reqResp = EventSourcedStreams.getProcessedCommands( ctx, commandRequestStream, commandResponseStream); final KStream<K, CommandRequest<K, C>> unprocessedRequests = reqResp.v1(); final KStream<K, CommandResponse<K>> processedResponses = reqResp.v2(); final KStream<K, CommandEvents<E, A>> commandEvents = EventSourcedStreams.getCommandEvents(ctx, unprocessedRequests, aggregateTable); final KStream<K, ValueWithSequence<E>> eventsWithSequence = EventSourcedStreams.getEventsWithSequence(commandEvents); final KStream<K, AggregateUpdateResult<A>> aggregateUpdateResults = EventSourcedStreams.getAggregateUpdateResults(ctx, commandEvents); final KStream<K, AggregateUpdate<A>> aggregateUpdates = EventSourcedStreams.getAggregateUpdates(aggregateUpdateResults); final KStream<K, CommandResponse<K>>commandResponses = EventSourcedStreams.getCommandResponses(aggregateUpdateResults); EventSourcedPublisher.publishEvents(ctx, eventsWithSequence); EventSourcedPublisher.publishAggregateUpdates(ctx, aggregateUpdates); EventSourcedPublisher.publishCommandResponses(ctx, processedResponses); EventSourcedPublisher.publishCommandResponses(ctx, commandResponses); ResultDistributor.distribute(distCtx, commandResponseStream, resultsTopicMapStream); return new InputStreams<>(commandRequestStream, commandResponseStream); }
EventSourcedTopology { public static <K, C, E, A> InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder) { final KStream<K, CommandRequest<K, C>> commandRequestStream = EventSourcedConsumer.commandRequestStream(ctx, builder); final KStream<K, CommandResponse<K>> commandResponseStream = EventSourcedConsumer.commandResponseStream(ctx, builder); final KTable<K, AggregateUpdate<A>> aggregateTable = EventSourcedConsumer.aggregateTable(ctx, builder); final DistributorContext<CommandId, CommandResponse<K>> distCtx = getDistributorContext(ctx); final KStream<CommandId, String> resultsTopicMapStream = ResultDistributor.resultTopicMapStream(distCtx, builder); Tuple2<KStream<K, CommandRequest<K, C>>, KStream<K, CommandResponse<K>>> reqResp = EventSourcedStreams.getProcessedCommands( ctx, commandRequestStream, commandResponseStream); final KStream<K, CommandRequest<K, C>> unprocessedRequests = reqResp.v1(); final KStream<K, CommandResponse<K>> processedResponses = reqResp.v2(); final KStream<K, CommandEvents<E, A>> commandEvents = EventSourcedStreams.getCommandEvents(ctx, unprocessedRequests, aggregateTable); final KStream<K, ValueWithSequence<E>> eventsWithSequence = EventSourcedStreams.getEventsWithSequence(commandEvents); final KStream<K, AggregateUpdateResult<A>> aggregateUpdateResults = EventSourcedStreams.getAggregateUpdateResults(ctx, commandEvents); final KStream<K, AggregateUpdate<A>> aggregateUpdates = EventSourcedStreams.getAggregateUpdates(aggregateUpdateResults); final KStream<K, CommandResponse<K>>commandResponses = EventSourcedStreams.getCommandResponses(aggregateUpdateResults); EventSourcedPublisher.publishEvents(ctx, eventsWithSequence); EventSourcedPublisher.publishAggregateUpdates(ctx, aggregateUpdates); EventSourcedPublisher.publishCommandResponses(ctx, processedResponses); EventSourcedPublisher.publishCommandResponses(ctx, commandResponses); ResultDistributor.distribute(distCtx, commandResponseStream, resultsTopicMapStream); return new InputStreams<>(commandRequestStream, commandResponseStream); } }
EventSourcedTopology { public static <K, C, E, A> InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder) { final KStream<K, CommandRequest<K, C>> commandRequestStream = EventSourcedConsumer.commandRequestStream(ctx, builder); final KStream<K, CommandResponse<K>> commandResponseStream = EventSourcedConsumer.commandResponseStream(ctx, builder); final KTable<K, AggregateUpdate<A>> aggregateTable = EventSourcedConsumer.aggregateTable(ctx, builder); final DistributorContext<CommandId, CommandResponse<K>> distCtx = getDistributorContext(ctx); final KStream<CommandId, String> resultsTopicMapStream = ResultDistributor.resultTopicMapStream(distCtx, builder); Tuple2<KStream<K, CommandRequest<K, C>>, KStream<K, CommandResponse<K>>> reqResp = EventSourcedStreams.getProcessedCommands( ctx, commandRequestStream, commandResponseStream); final KStream<K, CommandRequest<K, C>> unprocessedRequests = reqResp.v1(); final KStream<K, CommandResponse<K>> processedResponses = reqResp.v2(); final KStream<K, CommandEvents<E, A>> commandEvents = EventSourcedStreams.getCommandEvents(ctx, unprocessedRequests, aggregateTable); final KStream<K, ValueWithSequence<E>> eventsWithSequence = EventSourcedStreams.getEventsWithSequence(commandEvents); final KStream<K, AggregateUpdateResult<A>> aggregateUpdateResults = EventSourcedStreams.getAggregateUpdateResults(ctx, commandEvents); final KStream<K, AggregateUpdate<A>> aggregateUpdates = EventSourcedStreams.getAggregateUpdates(aggregateUpdateResults); final KStream<K, CommandResponse<K>>commandResponses = EventSourcedStreams.getCommandResponses(aggregateUpdateResults); EventSourcedPublisher.publishEvents(ctx, eventsWithSequence); EventSourcedPublisher.publishAggregateUpdates(ctx, aggregateUpdates); EventSourcedPublisher.publishCommandResponses(ctx, processedResponses); EventSourcedPublisher.publishCommandResponses(ctx, commandResponses); ResultDistributor.distribute(distCtx, commandResponseStream, resultsTopicMapStream); return new InputStreams<>(commandRequestStream, commandResponseStream); }  }
EventSourcedTopology { public static <K, C, E, A> InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder) { final KStream<K, CommandRequest<K, C>> commandRequestStream = EventSourcedConsumer.commandRequestStream(ctx, builder); final KStream<K, CommandResponse<K>> commandResponseStream = EventSourcedConsumer.commandResponseStream(ctx, builder); final KTable<K, AggregateUpdate<A>> aggregateTable = EventSourcedConsumer.aggregateTable(ctx, builder); final DistributorContext<CommandId, CommandResponse<K>> distCtx = getDistributorContext(ctx); final KStream<CommandId, String> resultsTopicMapStream = ResultDistributor.resultTopicMapStream(distCtx, builder); Tuple2<KStream<K, CommandRequest<K, C>>, KStream<K, CommandResponse<K>>> reqResp = EventSourcedStreams.getProcessedCommands( ctx, commandRequestStream, commandResponseStream); final KStream<K, CommandRequest<K, C>> unprocessedRequests = reqResp.v1(); final KStream<K, CommandResponse<K>> processedResponses = reqResp.v2(); final KStream<K, CommandEvents<E, A>> commandEvents = EventSourcedStreams.getCommandEvents(ctx, unprocessedRequests, aggregateTable); final KStream<K, ValueWithSequence<E>> eventsWithSequence = EventSourcedStreams.getEventsWithSequence(commandEvents); final KStream<K, AggregateUpdateResult<A>> aggregateUpdateResults = EventSourcedStreams.getAggregateUpdateResults(ctx, commandEvents); final KStream<K, AggregateUpdate<A>> aggregateUpdates = EventSourcedStreams.getAggregateUpdates(aggregateUpdateResults); final KStream<K, CommandResponse<K>>commandResponses = EventSourcedStreams.getCommandResponses(aggregateUpdateResults); EventSourcedPublisher.publishEvents(ctx, eventsWithSequence); EventSourcedPublisher.publishAggregateUpdates(ctx, aggregateUpdates); EventSourcedPublisher.publishCommandResponses(ctx, processedResponses); EventSourcedPublisher.publishCommandResponses(ctx, commandResponses); ResultDistributor.distribute(distCtx, commandResponseStream, resultsTopicMapStream); return new InputStreams<>(commandRequestStream, commandResponseStream); }  static InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder); }
EventSourcedTopology { public static <K, C, E, A> InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder) { final KStream<K, CommandRequest<K, C>> commandRequestStream = EventSourcedConsumer.commandRequestStream(ctx, builder); final KStream<K, CommandResponse<K>> commandResponseStream = EventSourcedConsumer.commandResponseStream(ctx, builder); final KTable<K, AggregateUpdate<A>> aggregateTable = EventSourcedConsumer.aggregateTable(ctx, builder); final DistributorContext<CommandId, CommandResponse<K>> distCtx = getDistributorContext(ctx); final KStream<CommandId, String> resultsTopicMapStream = ResultDistributor.resultTopicMapStream(distCtx, builder); Tuple2<KStream<K, CommandRequest<K, C>>, KStream<K, CommandResponse<K>>> reqResp = EventSourcedStreams.getProcessedCommands( ctx, commandRequestStream, commandResponseStream); final KStream<K, CommandRequest<K, C>> unprocessedRequests = reqResp.v1(); final KStream<K, CommandResponse<K>> processedResponses = reqResp.v2(); final KStream<K, CommandEvents<E, A>> commandEvents = EventSourcedStreams.getCommandEvents(ctx, unprocessedRequests, aggregateTable); final KStream<K, ValueWithSequence<E>> eventsWithSequence = EventSourcedStreams.getEventsWithSequence(commandEvents); final KStream<K, AggregateUpdateResult<A>> aggregateUpdateResults = EventSourcedStreams.getAggregateUpdateResults(ctx, commandEvents); final KStream<K, AggregateUpdate<A>> aggregateUpdates = EventSourcedStreams.getAggregateUpdates(aggregateUpdateResults); final KStream<K, CommandResponse<K>>commandResponses = EventSourcedStreams.getCommandResponses(aggregateUpdateResults); EventSourcedPublisher.publishEvents(ctx, eventsWithSequence); EventSourcedPublisher.publishAggregateUpdates(ctx, aggregateUpdates); EventSourcedPublisher.publishCommandResponses(ctx, processedResponses); EventSourcedPublisher.publishCommandResponses(ctx, commandResponses); ResultDistributor.distribute(distCtx, commandResponseStream, resultsTopicMapStream); return new InputStreams<>(commandRequestStream, commandResponseStream); }  static InputStreams<K, C> addTopology(TopologyContext<K, C, E, A> ctx, final StreamsBuilder builder);  }