Scaling reactive APIs - www.razorcodes.com

code1_Savingtheevents.java

@Component
@RequiredArgsConstructor
public class CustomerListener {

    private final CustomerRepository<CustomerUpdatedEvent> customerRepository;

    @StreamListener("customer-events")
    public void customerEventsConsumer(Flux<DataUpdatedEvent> event) {
        event.map(this::parseEvent)
             .doOnNext(MyService::doBusinessLogic)   // returns CustomerUpdatedEvent
             .subscribe(customerRepository::save);
    }
}