Блог

Я настроил, как мне кажется, относительно простую машину состояния пружины. Иногда (может быть, каждый раз) что-то вызывает внешнее событие, я получаю массу таких ошибок. Есть идеи, что я делаю не так?

Я получаю следующую ошибку примерно 10 раз подряд, когда это происходит.

 [parallel-2] ERROR reactor.core.publisher.Operators - Operator called default onErrorDropped reactor.core.Exceptions$ErrorCallbackNotImplemented: reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 10/10 Caused by: reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 10/10  at reactor.core.Exceptions.retryExhausted(Exceptions.java:290)  at reactor.util.retry.RetryBackoffSpec.lambda$static$0(RetryBackoffSpec.java:67)  at reactor.util.retry.RetryBackoffSpec.lambda$generateCompanion$4(RetryBackoffSpec.java:557)  at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:374)  at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerComplete(FluxConcatMap.java:295)  at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onComplete(FluxConcatMap.java:884)  at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)  at reactor.core.publisher.MonoFlatMap$FlatMapInner.onNext(MonoFlatMap.java:249)  at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:284)  at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187)  at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:232)  at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)  at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)  at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.complete(MonoIgnoreThen.java:284)  at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onNext(MonoIgnoreThen.java:187)  at reactor.core.publisher.MonoDelay$MonoDelayRunnable.propagateDelay(MonoDelay.java:270)  at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:285)  at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)  at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)  at java.base/java.util.concurrent.FutureTask.run(Unknown Source)  at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)  at java.base/java.lang.Thread.run(Unknown Source) Caused by: reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.  at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:56)  at org.springframework.statemachine.support.ReactiveStateMachineExecutor$1.lambda$null$0(ReactiveStateMachineExecutor.java:461)  at reactor.core.publisher.MonoCallable.subscribe(MonoCallable.java:56)  at reactor.core.publisher.FluxRetryWhen$RetryWhenMainSubscriber.resubscribe(FluxRetryWhen.java:215)  at reactor.core.publisher.FluxRetryWhen$RetryWhenOtherSubscriber.onNext(FluxRetryWhen.java:268)  at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:281)  at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:860)  at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1815)  

Моя государственная машина выглядит так:

  public abstract class DBStateMachine extends StateMachineConfigurerAdapterlt;DBStates, DBStateChangesgt; {   public abstract Guardlt;DBStates, DBStateChangesgt; isDown();   public abstract Guardlt;DBStates, DBStateChangesgt; isReady();   @Override  public void configure(StateMachineStateConfigurerlt;DBStates, DBStateChangesgt; states)  throws Exception {   states  .withStates()  .initial(DBStates.INITIALIZING)  .junction(DBStates.TESTING)  // .end(DBStates.READY)  .states(EnumSet.allOf(DBStates.class));   }   @Override  public void configure(  StateMachineTransitionConfigurerlt;DBStates, DBStateChangesgt; transitions)  throws Exception {  //@formatter:off  transitions.withInternal()  .source(DBStates.INITIALIZING)  .action(timerAction())  .timer(100).and()  .withJunction()  .source(DBStates.TESTING)  .first(DBStates.DOWN, isDown())  .then(DBStates.READY, isReady())  .last(DBStates.BAD_PASSWORD).and()  .withExternal()  .source(DBStates.BAD_PASSWORD)  .target(DBStates.TESTING)  .event(DBStateChanges.PASSWORD_RECEIVED).and()  .withInternal()  .source(DBStates.DOWN)  .action(timerAction())  .timer(10000).and()  .withExternal()  .source(DBStates.DOWN)  .target(DBStates.TESTING)  .event(DBStateChanges.RETRY).and()  .withExternal()  .source(DBStates.INITIALIZING)  .target(DBStates.TESTING)  .event(DBStateChanges.RETRY).and()  .withExternal()  .source(DBStates.READY)  .target(DBStates.TESTING)  .event(DBStateChanges.ERROR);  //@formatter:on  }   @Bean  public TimerAction timerAction() {  return new TimerAction();  }   public class TimerAction implements Actionlt;DBStates, DBStateChangesgt; {   @Override  public void execute(StateContextlt;DBStates, DBStateChangesgt; context) {  // do something in every 1 sec  // context.getStateMachine().  // TODO make this reactive?  context.getStateMachine().sendEvent(DBStateChanges.RETRY);  }  }     }  

Я думаю, что код, вызывающий ошибку, заключается в следующем:

 @Autowired  private StateMachinelt;DBStates, DBStateChangesgt; statemachine; public void triggerStateChange() {  statemachine.sendEvent(DBStateChanges.ERROR);  }  

Редактировать

Я также изменил метод triggerStateChange, чтобы он выглядел так, и все равно получал ошибку.

 public void triggerStateChange() {    Messagelt;DBStateChangesgt; m = new GenericMessagelt;gt;(DBStateChanges.ERROR);   statemachine.sendEvent(Mono.just(m)).subscribe(); // statemachine.sendEvent(DBStateChanges.ERROR);  }  

Я должен упомянуть, что я на Java 16.0.1, Spring Boot 2.5, Spring Framework 5.3.7 и Spring State Machine 3.0.1.