Page History

final MessageRouterPublisher publisher = DmaapClientFactory.createMessageRouterPublisher();
final MessageRouterSink sinkDefinition; //... Sink definition obtained by parsing CBS response
final MessageRouterPublishRequest request = ImmutableMessageRouterPublishRequest.builder()
        .sinkDefinition(sinkDefinition)
        .build();

Flux.just(1, 2, 3)
        .map(JsonPrimitive::new)
        .transform(input -> cut.put(request, input))
        .subscribe(resp -> {
                    if (resp.successful()) {
                        logger.debug("Sent a batch of messages to the MR");
                    } else {
                        logger.warn("Message sending has failed: {}", resp.failReason());
                    }
                },
                ex -> {
                    logger.warn("An unexpected error while sending messages to DMaaP", ex);
                });