@Test void testDistributor() { String topicNamesTopic = "topic_names"; String outputTopic = "output_topic"; TopologyContext<String, TestCommand, TestEvent, Optional<TestAggregate>> ctx = ctxBuilder.buildContext(); driver = new TestDriverInitializer().build(builder -> { EventSourcedTopology.InputStreams<String, TestCommand> inputStreams = EventSourcedTopology.addTopology(ctx, builder); DistributorContext<CommandId, CommandResponse<String>> context = new DistributorContext<>( topicNamesTopic, new DistributorSerdes<>(ctx.serdes().commandId(), ctx.serdes().commandResponse()), ctx.aggregateSpec().generation().stateStoreSpec(), CommandResponse::commandId, CommandId::id); KStream<CommandId, String> topicNames = builder.stream(topicNamesTopic, Consumed.with(ctx.serdes().commandId(), Serdes.String())); ResultDistributor.distribute(context, inputStreams.commandResponse, topicNames); }); TestContextDriver<String, TestCommand, TestEvent, Optional<TestAggregate>> ctxDriver = new TestContextDriver<>(ctx, driver); CommandRequest<String, TestCommand> commandRequest = CommandRequest.of( CommandId.random(), key, Sequence.first(), new TestCommand.CreateCommand("Name 2")); ctxDriver.getPublisher(ctx.serdes().commandId(), Serdes.String()) .publish(topicNamesTopic, commandRequest.commandId(), outputTopic); ctxDriver.publishCommand( key, commandRequest); ProducerRecord<String, CommandResponse<String>> output = driver.readOutput(outputTopic, Serdes.String().deserializer(), ctx.serdes().commandResponse().deserializer()); assertThat(output.key()).isEqualTo(String.format("%s:%s", outputTopic, commandRequest.commandId().id().toString())); assertThat(output.value().sequenceResult().isSuccess()).isEqualTo(true); }
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);  }