When the AddNotification command is sent
commandGateway.send<Unit>(
RequestConfirmPayer(
externalId = externalId,
rqUID = it.rqUID,
phoneNumber = it.phoneNumber)
).then(getProjectionProducedByConfirmPayerId(externalId))
.map { confirmPayerRs ->
commandGateway.send<Unit>(
AddNotification(
requestNumber = event.requestNumber,
transactionStatus = confirmPayerRs.transactionStatus,
amount = confirmPayerRs.amount)
).block()
}.subscribe()
private fun getProjectionProducedByConfirmPayerId(externalId: ExternalId): Mono<ConfirmPayerRs> {
return queryGateway.subscriptionQuery(GetConfirmPayer(externalId), ResponseTypes.instanceOf(Void::class.java), ResponseTypes.instanceOf(ConfirmPayerRs::class.java)
).flatMap { queryResult ->
queryResult.updates().next().timeout(Duration.ofSeconds(20)).doFinally { queryResult.close() }
}
}
I have the below error
java.lang.IllegalStateException: It is not allowed to publish events when the root Unit of Work has already been committed.
at org.axonframework.common.Assert.state(Assert.java:44) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.AbstractEventBus.publish(AbstractEventBus.java:132) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.EventBus.publish(EventBus.java:50) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.publishOnEventBus(AnnotatedAggregate.java:389) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.eventsourcing.EventSourcedAggregate.publishOnEventBus(EventSourcedAggregate.java:266) ~[axon-eventsourcing-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.publish(AnnotatedAggregate.java:379) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.eventsourcing.EventSourcedAggregate.publish(EventSourcedAggregate.java:253) ~[axon-eventsourcing-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.doApply(AnnotatedAggregate.java:463) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.eventsourcing.EventSourcedAggregate.doApply(EventSourcedAggregate.java:247) ~[axon-eventsourcing-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.AggregateLifecycle.apply(AggregateLifecycle.java:62) ~[axon-modelling-4.6.1.jar:4.6.1]
at com.hts.payment.command.PaymentAggregate.handle(Aggregate.kt:118) ~[classes/:na]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) ~[na:na]
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
at java.base/java.lang.reflect.Method.invoke(Method.java:568) ~[na:na]
at org.axonframework.messaging.annotation.AnnotatedMessageHandlingMember.handle(AnnotatedMessageHandlingMember.java:153) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.annotation.WrappedMessageHandlingMember.handle(WrappedMessageHandlingMember.java:64) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.annotation.WrappedMessageHandlingMember.handle(WrappedMessageHandlingMember.java:64) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.TracingHandlerEnhancerDefinition$1.lambda$handle$1(TracingHandlerEnhancerDefinition.java:84) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.Span.runCallable(Span.java:103) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.TracingHandlerEnhancerDefinition$1.handle(TracingHandlerEnhancerDefinition.java:84) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.findHandlerAndHandleCommand(AnnotatedAggregate.java:450) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.handle(AnnotatedAggregate.java:426) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.lambda$handle$3(AnnotatedAggregate.java:398) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.messaging.Scope.executeWithResult(Scope.java:111) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.inspection.AnnotatedAggregate.handle(AnnotatedAggregate.java:405) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.LockAwareAggregate.handle(LockAwareAggregate.java:97) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.AggregateAnnotationCommandHandler$AggregateCommandHandler.handle(AggregateAnnotationCommandHandler.java:561) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.modelling.command.AggregateAnnotationCommandHandler$AggregateCommandHandler.handle(AggregateAnnotationCommandHandler.java:550) ~[axon-modelling-4.6.1.jar:4.6.1]
at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:57) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.interceptors.LoggingInterceptor.handle(LoggingInterceptor.java:85) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.interceptors.CorrelationDataInterceptor.handle(CorrelationDataInterceptor.java:67) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.DefaultInterceptorChain.proceed(DefaultInterceptorChain.java:55) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.executeWithResult(DefaultUnitOfWork.java:77) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.commandhandling.SimpleCommandBus.lambda$handle$3(SimpleCommandBus.java:198) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.Span.run(Span.java:72) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.commandhandling.SimpleCommandBus.handle(SimpleCommandBus.java:189) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.commandhandling.SimpleCommandBus.doDispatch(SimpleCommandBus.java:163) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.commandhandling.SimpleCommandBus.dispatch(SimpleCommandBus.java:130) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.extensions.reactor.commandhandling.gateway.DefaultReactorCommandGateway.lambda$dispatchCommand$5(DefaultReactorCommandGateway.java:118) ~[axon-reactor-4.6.0.jar:4.6.0]
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2398) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:171) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMap$FlatMapInner.onSubscribe(MonoFlatMap.java:238) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:96) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoDeferContextual.subscribe(MonoDeferContextual.java:55) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoReduceSeed$ReduceSeedSubscriber.onComplete(MonoReduceSeed.java:165) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Operators.complete(Operators.java:137) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:148) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:87) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Mono.subscribe(Mono.java:4455) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Mono.block(Mono.java:1706) ~[reactor-core-3.4.23.jar:3.4.23]
at com.hts.payment.PaymentProcess.handle$lambda-12$lambda-11(Process.kt:163) ~[classes/:na]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:106) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:292) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onNext(FluxDoFinally.java:113) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.SerializedSubscriber.onNext(SerializedSubscriber.java:99) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.SerializedSubscriber.onNext(SerializedSubscriber.java:99) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.onNext(FluxTimeout.java:180) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:82) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxFilter$FilterSubscriber.onNext(FluxFilter.java:113) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.drain(FluxMergeSequential.java:432) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.innerComplete(FluxMergeSequential.java:328) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMergeSequential$MergeSequentialInner.onSubscribe(FluxMergeSequential.java:552) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:68) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.Flux.subscribe(Flux.java:8522) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMergeSequential$MergeSequentialMain.onNext(FluxMergeSequential.java:237) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:877) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay(FluxReplay.java:965) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxReplay$ReplaySubscriber.onNext(FluxReplay.java:1344) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:537) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:343) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:309) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxCreate$NoOverflowBaseAsyncSink.next(FluxCreate.java:669) ~[reactor-core-3.4.23.jar:3.4.23]
at reactor.core.publisher.FluxCreate$SerializedFluxSink.next(FluxCreate.java:161) ~[reactor-core-3.4.23.jar:3.4.23]
at org.axonframework.queryhandling.FluxSinkWrapper.next(FluxSinkWrapper.java:60) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.lambda$doEmit$15(SimpleQueryUpdateEmitter.java:221) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.Span.run(Span.java:72) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.doEmit(SimpleQueryUpdateEmitter.java:217) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.lambda$null$11(SimpleQueryUpdateEmitter.java:194) ~[axon-messaging-4.6.1.jar:4.6.1]
at java.base/java.util.Optional.ifPresent(Optional.java:178) ~[na:na]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.lambda$doEmit$12(SimpleQueryUpdateEmitter.java:194) ~[axon-messaging-4.6.1.jar:4.6.1]
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[na:na]
at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179) ~[na:na]
at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179) ~[na:na]
at java.base/java.util.concurrent.ConcurrentHashMap$KeySpliterator.forEachRemaining(ConcurrentHashMap.java:3573) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[na:na]
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[na:na]
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:na]
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) ~[na:na]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.doEmit(SimpleQueryUpdateEmitter.java:193) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.lambda$emit$6(SimpleQueryUpdateEmitter.java:155) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.Span.run(Span.java:72) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.tracing.Span.lambda$wrapRunnable$0(Span.java:90) ~[axon-messaging-4.6.1.jar:4.6.1]
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511) ~[na:na]
at org.axonframework.queryhandling.SimpleQueryUpdateEmitter.lambda$null$20(SimpleQueryUpdateEmitter.java:287) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.MessageProcessingContext.notifyHandlers(MessageProcessingContext.java:72) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.lambda$notifyHandlers$2(BatchingUnitOfWork.java:161) ~[axon-messaging-4.6.1.jar:4.6.1]
at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133) ~[na:na]
at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.notifyHandlers(BatchingUnitOfWork.java:161) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.changePhase(AbstractUnitOfWork.java:236) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commitAsRoot(AbstractUnitOfWork.java:94) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commit(AbstractUnitOfWork.java:75) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.executeWithResult(BatchingUnitOfWork.java:117) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.AbstractEventProcessor.processInUnitOfWork(AbstractEventProcessor.java:166) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.TrackingEventProcessor.processBatch(TrackingEventProcessor.java:490) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.TrackingEventProcessor.processingLoop(TrackingEventProcessor.java:318) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.TrackingEventProcessor$TrackingSegmentWorker.run(TrackingEventProcessor.java:1145) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.cleanUp(TrackingEventProcessor.java:1340) ~[axon-messaging-4.6.1.jar:4.6.1]
at org.axonframework.eventhandling.TrackingEventProcessor$WorkerLauncher.run(TrackingEventProcessor.java:1317) ~[axon-messaging-4.6.1.jar:4.6.1]
at java.base/java.lang.Thread.run(Thread.java:833) ~[na:an]
