Page Comparison

final MessageRouterSource sourceDefinition; //... Source definition obtained by parsing CBS response
final MessageRouterSubscribeRequest request = ImmutableMessageRouterPublishRequest.builder()
        .sinkDefinition(sourceDefinition)
        .build();

cut.subscribeForElements(request, Duration.ofMinutes(1))
        .map(JsonElement::getAsJsonObject)
        .doOnNextsubscribe(json -> {
                // application logic
            }),
            ex -> {
                logger.subscribe(warn("An unexpected error while receiving messages from DMaaP", ex);
            });