Cloud Foundry Java Client - Streaming events

Cloud Foundry Java Client provides Java based bindings for interacting with a running Cloud Foundry instance. One of the neat things about this project is that it has embraced the Reactive Stream based API's for its method signatures, specifically using the Reactor implementation, this is especially useful when consuming streaming data.

In this post I want to demonstrate a specific use case where this library really shines - in streaming events from Cloud Foundry

Loggregator is the subsystem in Cloud Foundry responsible for aggregating all the logs produced within the system and provides ways for this information to be streamed out to external systems. The "Traffic Controller" component within Loggregator exposes a Websocket based endpoint streaming out these events, the Cloud Foundry Java client abstracts the underlying websocket client connection details and provides a neat way to consume this information.


As a pre-requisite, you will need a running instance of Cloud Foundry to try out the sample and the best way to get it working locally is to use PCF Dev.

Assuming that you have a running instance, the way to connect to this instance from code using the cf-java-client library is along the following lines:

SpringCloudFoundryClient cfClient = SpringCloudFoundryClient.builder()
                .host("api.local.pcfdev.io")
                .username("admin")
                .password("admin")
                .skipSslValidation(true)
                .build();

Using this, a client to the Traffic Controller can be created the following way:

DopplerClient dopplerClient = ReactorDopplerClient.builder()
      .cloudFoundryClient(cfClient)
      .build();

That is essentially it, doppler client provides methods to stream the underlying events, if you are interested in all the unfiltered information(appropriately referred to as the firehose), you can do it the following way:

Flux<Event> cfEvents = this.dopplerClient.firehose(
    FirehoseRequest.builder()
      .subscriptionId(UUID.randomUUID().toString()).build());

The result is a Flux type from the Reactor library encapsulating the streaming data which can be observed by attaching a subscriber, say for a basic example of a subscriber simply logging the events to the console the following way:

cfEvents.subscribe(e -> LOGGER.info(e.toString()));

However the real power of Flux is in the very powerful fluent methods that it provides, so for eg if I were interested in a subset of say just the Application level logs, I would essentially want to filter down the data, extract the log from it and print the log the following way:

cfEvents
 .filter(e -> LogMessage.class.isInstance(e))
 .map(e -> (LogMessage)e)
 .map(LogMessage::getMessage)
 .subscribe(LOGGER::info);

If you want to play with this sample which as an added bonus has been Spring Boot enabled, I have it available in my github repository.