...
- You can extract raw streams by means of
DataStreams.namedSinks(for streams_publishes) andDataStreams.namedSources(for streams_subscribes). - Then you will be able to parse the specific entry from returned collection to a desired stream type by means of parsers built by
StreamFromGsonParsersfactory.
Sample usage:
| Code Block | ||||
|---|---|---|---|---|
| ||||
final CbsRequest request = CbsRequests.getConfiguration(RequestDiagnosticContext.create()); final StreamFromGsonParser<MessageRouterSink> mrSinkParser = StreamFromGsonParsers.messageRouterSinkParser(); CbsClientFactory.createCbsClient(EnvProperties.fromEnvironment()) .flatMapMany(cbsClient -> cbsClient.updates(request, Duration.ofSeconds(5), Duration.ofMinutes(1))) .map(DataStreams::namedSinks) .map(sinks -> sinks.filter(StreamPredicates.streamOfType(MESSAGE_ROUTER)).map(mrSinkParser::unsafeParse).toList()) .subscribe( mrSinks -> mrSinks.forEach(mrSink -> { logger.info(mrSink.name()); logger.info(mrSink.aafCredentials().username()); logger.info(mrSink.topicUrl()); // ... }), throwable -> logger.warn("Ooops", throwable) ); |
For details and sample usage please refer to JavaDoc and unit and integration tests. Especially CbsClientImplIT and MixedDmaapStreamsIT might be usefull.
...
crypt-password - an utility for BCrypt passwords
...