Пример реактора Eventbus, который возвращает ответ или сообщение об ошибке

Question

Прежде всего, я довольно новичок в области асинхронной обработки. В моем текущем проекте мы используем весенний ботинок вместе с проектным реактором, в частности Eventbus, для выполнения асинхронной обработки. Использование eventbus, я думаю, также сделает нашу систему более масштабируемой.

До сих пор использование EventBus было довольно ограничено, когда мы выполняем некоторую обработку у потребителя EventBus, который ничего не возвращает. Configiuration и пример процессор заключается в следующем:

// Config File

@SpringBootApplication 
public class Application implements CommandLineRunner { 

@Autowired 
private EventBus eventBus; 

@Autowired 
private BatchProcessor batchProcessor; 

@Override 
public void run(String... arg0) throws Exception { 
    eventBus.on("batchProcessor", batchProcessor); 

} 
} 

// Потребительские

@Service 
public class BatchProcesspr implements Consumer<Event<Request>> { 

@Override 
    public void accept(Event<Request> event) { 
     // processing goes here 
} 

До сих пор это было хорошо с помощью метода принимает имеющий тип возвращаемого значения аннулируется. Но теперь у меня есть сценарий, в котором я хотел бы вернуть ответ от метода процессора или если возникла ошибка, а обработка должна была вызвать соответствующее исключение, и в любом случае ответ/исключение должно быть возвращено в точку призывание.

Можно ли это сделать, используя реактор? если да, пожалуйста, дайте простой пример для этого. Я читал о Promise, но не могу найти пример, похожий на мой случай.

Answer 1

Вы попробовали sendAndReceive? http://projectreactor.io/ext/docs/reference/#bus-request-reply

EventBus bus; 

bus.receive($("job.sink"), (Event<String> ev) -> { 
    return ev.getData().toUpperCase(); 
}); 

bus.sendAndReceive(
    "job.sink", 
    Event.wrap("Hello World!"), 
    s -> System.out.printf("Got %s on thread %s%n", s, Thread.currentThread()) 
); 

Вы можете легко зарегистрировать другой потребитель на стороне вызывающего абонента, который уведомлен, когда служба отвечает.