AckRejectionPublisher implements StreamObserver<Ack> { @Override public void onNext(Ack value) { checkNotNull(value); Status status = value.getStatus(); if (status.getStatusCase() == REJECTION) { eventBus.post(status.getRejection()); } } AckRejectionPublisher(EventBus eventBus); @Override void onNext(Ack value); @Override void onError(Throwable t); @Override void onCompleted();  }
@Test @DisplayName("publish the passed rejection to event bus") void publishRejection() { TestActorRequestFactory requestFactory = new TestActorRequestFactory(AckRejectionPublisherTest.class); Command command = requestFactory.generateCommand(); CommandEnvelope origin = CommandEnvelope.of(command); CmdBusEntryDenied throwableMessage = CmdBusEntryDenied .newBuilder() .setId(CmdBusCaffetteriaId.generate()) .setVisitorCount(15) .setReason("Test command bus rejection.") .build(); RejectionEnvelope rejection = RejectionEnvelope.from(origin, throwableMessage); Ack ack = Acks.reject(CommandId.generate(), rejection); publisher.onNext(ack); assertThat(listener.lastReceived).isNotNull(); assertThat(listener.lastReceived.message()).isEqualTo(throwableMessage.messageThrown()); }
@Test @DisplayName("ignore non-rejection `Ack`s") void ignoreNonRejection() { Ack ack = Acks.acknowledge(CommandId.generate()); publisher.onNext(ack); assertThat(listener.lastReceived).isNull(); }