Because most services and collectors deployed on DCAE platform relies on similar microservices a commmon Software Development Kit has been created. It contains utilities and clients which may be used when getting configuration from CBS, consuming messages from DMaaP, interacting with AAI, etc. SDK is written in Java.
NOTE For now only Config Binding Service client has reached a stable API level. More are expected to be done in Dublin.
NOTE This page is a work in progress.
Table of Contents |
---|
Reactive programming
Most of SDK APIs are using Project Reactor, which is one of available implementation of Reactive Streams (as well as Java 9 Flow). This way we support both high-performance, non-blocking asynchronous clients and old-school, thread-bound, blocking clients. We believe that using reactive programming can solve many cloud-specific problems for us - if used properly.
Handling streams of data—especially “live” data whose volume is not predetermined—requires special care in an asynchronous system. The most prominent issue is that resource consumption needs to be controlled such that a fast data source does not overwhelm the stream destination. Asynchrony is needed in order to enable the parallel use of computing resources, on collaborating network hosts or multiple CPU cores within a single machine.
The main goal of Reactive Streams is to govern the exchange of stream data across an asynchronous boundary—think passing elements on to another thread or thread-pool—while ensuring that the receiving side is not forced to buffer arbitrary amounts of data. In other words, back pressure is an integral part of this model in order to allow the queues which mediate between threads to be bounded. The benefits of asynchronous processing would be negated if the communication of back pressure were synchronous (see also the Reactive Manifesto), therefore care has to be taken to mandate fully non-blocking and asynchronous behavior of all aspects of a Reactive Streams implementation.
Before using DCAE SDK, please take a moment and read Project Reactor documentation. You should also skim through methods available in Flux and Mono. It can save you many wasted minutes/hours of thinking how to implement the functionality by your own instead of using already implemented method.
Some general notes:
- Do not block in any of handlers which are passed to operators defined by Reactor. The library uses a set of Schedulers (think thread-pools) which are suitable for different jobs. More details can be found in the documentation. If possible try to use non-blocking APIs.
- Most of operators support back-pressure. That means that a demand for new messages will be signalized from downstream subscribers. For instance if you have a
flux.flatMap(this::doThis).map(this::doThat).subscribe()
then ifdoThis
is very slow it will not request many items from sourceflux
and it will emit items at it's own pace fordoThat
to process. So usually there will be no buffering nor blocking needed betweenflux
anddoThis
. - (Almost) nothing will happen without subscribing to the Flux/Mono. These reactive streams are lazy, so the demand will be signaled only when subscription is being made ie. by means of
subscribe
orblock*
methods. - If you are going to go fully-reactive then you should probably not call subscribe/block anywhere in your code. For instance, when using Reactor Netty or Spring Web Flux you should return Mono/Flux from your core methods and it will be subscribed somewhere by the library you are using.
However if you are not concerned about performance you can always take a blue pill and go back to blocking world by means of block*
methods.
Artifacts
Current version
Code Block | ||
---|---|---|
| ||
<properties>
<sdk.version>1.1.2-SNAPSHOT</sdk.version>
</properties> |
Maven dependencies
Choose one or more depending on what you need.
Code Block | ||
---|---|---|
| ||
<dependencies>
<dependency>
<groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
<artifactId>cbs-client</artifactId>
<version>${sdk.version}</version>
</dependency>
<!-- more to go -->
</dependencies> |
Available APIs
cbs-client
- a Config Binding Service client
You can use CbsClientFactory to lookup for CBS in your application. Returned CbsClient can then be used to get a configuration, poll for configuration or poll for configuration changes.
Sample usage:
Code Block | ||||
---|---|---|---|---|
| ||||
// Generate RequestID and InvocationID which will be used when logging and in HTTP requests
RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
// Read necessary properties from the environment
final EnvProperties env = EnvProperties.fromEnvironment();
// Create the client and use it to get the configuration
CbsClientFactory.createCbsClient(env)
.flatMap(cbsClient -> cbsClient.get(diagnosticContext))
.subscribe(
jsonObject -> {
// do a stuff with your JSON configuration using GSON API
final int port = Integer.parseInt(jsonObject.get("collector.listen_port").getAsString());
// ...
},
throwable -> {
logger.warn("Ooops", throwable);
}); |
Note that a subscribe handler can/will be called in separate thread asynchronously after CBS address lookup succeeds and CBS service call returns a result.
If you are interested in calling CBS periodically and react only when the configuration changed you can use updates
method:
Code Block | ||||
---|---|---|---|---|
| ||||
// Generate RequestID and InvocationID which will be used when logging and in HTTP requests
RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
// Read necessary properties from the environment
final EnvProperties env = EnvProperties.fromEnvironment();
// Polling properties
final Duration initialDelay = Duration.ofSeconds(5);
final Duration period = Duration.ofMinutes(1);
// Create the client and use it to get the configuration
CbsClientFactory.createCbsClient(env)
.flatMapMany(cbsClient -> cbsClient.updates(diagnosticContext, initialDelay, period))
.subscribe(
jsonObject -> {
// do a stuff with your JSON configuration using GSON API
final int port = Integer.parseInt(jsonObject.get("collector.listen_port").getAsString());
// ...
},
throwable -> {
logger.warn("Ooops", throwable);
}); |
Click here to see SDK architecture description.
Artifacts
Click here to see artifacts description.
Available APIs
Click here to see available APIs descriptionThe most significant change is in line 13. We are using flatMapMany since we want to map one CbsClient to many JsonObject updates. After 5 seconds CbsClient will call CBS every minute. If the configuration has changed it will pass the JsonObject downstream - in our case consumer of JsonObject will be called.