@Test void successfulCommandResponse() { 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); String name = "name"; CommandRequest<String, TestCommand> commandRequest = CommandRequest.of( CommandId.random(), key, Sequence.first(), new TestCommand.CreateCommand(name)); ctxDriver.publishCommand( key, commandRequest); ctxDriver.verifyCommandResponse(key, v -> { assertThat(v.sequenceResult().isSuccess()).isEqualTo(true); assertThat(v.sequenceResult().getOrElse(Sequence.position(1000)).getSeq()).isEqualTo(Sequence.first().next().getSeq()); }); ctxDriver.verifyNoCommandResponse(); ctxDriver.verifyEvent(key, e -> { assertThat(e.value()).isEqualTo(new TestEvent.Created(name)); assertThat(e.sequence().getSeq()).isEqualTo(Sequence.first().next().getSeq()); }); ctxDriver.verifyNoEvent(); ctxDriver.verifyAggregateUpdate(key, a -> { assertThat(a.sequence().getSeq()).isEqualTo(1L); assertThat(a.aggregate().get()).isEqualTo(new TestAggregate(name)); }); 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);  }