Hot and cold rx-java Observable

My own understanding of Hot and Cold Observable is quite shaky, but here is what I have understood till now!

Cold Observable

Consider an API which returns an rx-java Observable:

import obs.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.schedulers.Schedulers;

public class Service1 {
    private static final Logger logger = LoggerFactory.getLogger(Service1.class);
    public Observable<String> operation() {
        return Observable.<String>create(s -> {
            logger.info("Start: Executing slow task in Service 1");
            Util.delay(1000);
            s.onNext("data 1");
            logger.info("End: Executing slow task in Service 1");
            s.onCompleted();
        }).subscribeOn(Schedulers.computation());
    }
}

Now, the first thing to note is that the typical Observable does not do anything until it is subscribed to:

So essentially if I were to do this:

Observable<String> op1 = service1.operation();

Nothing would be printed or returned, unless there is a subscription on the Observable this way:

Observable<String> op1 = service1.operation();

CountDownLatch latch = new CountDownLatch(1);

op1.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

latch.await();

So now, what happens if there are multiple subscriptions on this Observable:

Observable<String> op1 = service1.operation();

CountDownLatch latch = new CountDownLatch(3);

op1.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

op1.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

op1.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

latch.await();

With a cold observable the code would get called once more and the items emitted again, I get this on my machine:

06:04:07.206 [RxComputationThreadPool-2] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:04:07.208 [RxComputationThreadPool-3] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:04:08.211 [RxComputationThreadPool-2] INFO  o.b.BasicObservablesTest - From Subscriber 2: data 1
06:04:08.211 [RxComputationThreadPool-1] INFO  o.b.BasicObservablesTest - From Subscriber 1: data 1
06:04:08.211 [RxComputationThreadPool-3] INFO  o.b.BasicObservablesTest - From Subscriber 3: data 1
06:04:08.213 [RxComputationThreadPool-2] INFO  o.b.Service1 - End: Executing slow task in Service 1
06:04:08.214 [RxComputationThreadPool-1] INFO  o.b.Service1 - End: Executing slow task in Service 1
06:04:08.214 [RxComputationThreadPool-3] INFO  o.b.Service1 - End: Executing slow task in Service 1

Hot Observable - using ConnectableObservable

Hot Observable on the other hand does not really need a subscription to start emitting items. A way to implement a Hot Observable is using a ConnectableObservable, which is a Observable which does not emit items until its connect method is called, however once it starts emitting items, any subscriber to it gets items only from the point of subscription. So again revisiting the previous example, but with a ConnectableObservable instead:

Observable<String> op1 = service1.operation();

ConnectableObservable<String> connectableObservable =  op1.publish();

CountDownLatch latch = new CountDownLatch(3);

connectableObservable.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.connect();

latch.await();

and the following gets printed:

06:07:23.852 [RxComputationThreadPool-3] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:07:24.860 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 1: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 2: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 3: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.Service1 - End: Executing slow task in Service 1

Hot Observable - using Subject

Another way to convert a cold Observable to a hot one is to use a Subject. Subjects behave both as an Observable and an Observer, there are different types of Subjects available with different behavior. Here I am using a Subject called a PublishSubject which has a Pub/Sub behavior - the items get emitted to all the subscribers listening on it. So with a PublishSubject introduced the code looks like this:

Observable<String> op1 = service1.operation();

PublishSubject<String> publishSubject = PublishSubject.create();

op1.subscribe(publishSubject);

CountDownLatch latch = new CountDownLatch(3);

publishSubject.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

publishSubject.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

publishSubject.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());


latch.await();

See how the PublishSubject is introduced as a subscriber to the Observable and the other subscribers subscribe to the PublishSubject instead. The output will be similar to the one from ConnectableObservable.

This is essentially it, the extent of my understanding of Hot Observable. So to conclude, the difference between a Cold and a Hot Observable is about when the subscribers get the emitted items and when the items are emitted - with a Cold Observable they are emitted when they are subscribed to and typically get all the emitted items, with a Hot Observable the items are emitted without a Subscriber and subscribers get items emitted after the point of subscription typically.

Reference

1. http://www.introtorx.com/content/v1.0.10621.0/14_HotAndColdObservables.html
2. Excellent javadoc on rx-java - http://reactivex.io/RxJava/javadoc/index.html

Using rx-java Observable in a Spring MVC flow

Spring MVC has supported asynchronous request processing flow for sometime now and this support internally utilizes the Servlet 3 async support of containers like Tomcat/Jetty.

Spring Web Async support

Consider a service call that takes a little while to process, simulated with a delay:

public CompletableFuture<Message> getAMessageFuture() {
    return CompletableFuture.supplyAsync(() -> {
        logger.info("Start: Executing slow task in Service 1");
        Util.delay(1000);
        logger.info("End: Executing slow task in Service 1");
        return new Message("data 1");
    }, futureExecutor);
}

If I were to call this service in a user request flow, the traditional blocking controller flow would look like this:

@RequestMapping("/getAMessageFutureBlocking")
public Message getAMessageFutureBlocking() throws Exception {
    return service1.getAMessageFuture().get();
}

A better approach is to use the Spring Asynchronous support to return the result back to the user when available from the CompletableFuture, this way not holding up the containers thread:

@RequestMapping("/getAMessageFutureAsync")
public DeferredResult<Message> getAMessageFutureAsync() {
    DeferredResult<Message> deffered = new DeferredResult<>(90000);
    CompletableFuture<Message> f = this.service1.getAMessageFuture();
    f.whenComplete((res, ex) -> {
        if (ex != null) {
            deffered.setErrorResult(ex);
        } else {
            deffered.setResult(res);
        }
    });
    return deffered;
}

Using Observable in a Async Flow

Now to the topic of this article, I have been using Rx-java's excellent Observable type as my service return types lately and wanted to ensure that the web layer also remains asynchronous in processing the Observable type returned from a service call.

Consider the service that was described above now modified to return an Observable:

public Observable<Message> getAMessageObs() {
    return Observable.<Message>create(s -> {
        logger.info("Start: Executing slow task in Service 1");
        Util.delay(1000);
        s.onNext(new Message("data 1"));
        logger.info("End: Executing slow task in Service 1");
        s.onCompleted();
    }).subscribeOn(Schedulers.from(customObservableExecutor));
}

I can nullify all the benefits of returning an Observable by ending up with a blocking call at the web layer, a naive call will be the following:

@RequestMapping("/getAMessageObsBlocking")
public Message getAMessageObsBlocking() {
    return service1.getAMessageObs().toBlocking().first();
}

To make this flow async through the web layer, a better way to handle this call is the following, essentially by transforming Observable to Spring's DeferredResult type:

@RequestMapping("/getAMessageObsAsync")
public DeferredResult<Message> getAMessageAsync() {
    Observable<Message> o = this.service1.getAMessageObs();
    DeferredResult<Message> deffered = new DeferredResult<>(90000);
    o.subscribe(m -> deffered.setResult(m), e -> deffered.setErrorResult(e));
    return deffered;
}

This would ensure that the thread handling the user flow would return as soon as the service call is complete and the user response will be processed reactively once the observable starts emitting values.


If you are interested in exploring this further, here is a github repo with working samples: https://github.com/bijukunjummen/spring-web-observable.

References:

Spring's reference guide on async flows in the web tier: http://docs.spring.io/spring/docs/current/spring-framework-reference/html/mvc.html#mvc-ann-async

More details on Spring DeferredResult by the inimitable Tomasz Nurkiewicz at the NoBlogDefFound blog - http://www.nurkiewicz.com/2013/03/deferredresult-asynchronous-processing.html

Netflix Archaius properties in a Spring project

Archaius Basics

Netflix Archaius is a library for managing configuration for an application. Consider a properties file "sample.properties" holding a property called "myprop":

myprop=myprop_value_default

This is how the file is loaded up using Archaius:

ConfigurationManager
                .loadCascadedPropertiesFromResources("sample");

String myProp = DynamicPropertyFactory.getInstance().getStringProperty("myprop", "NOT FOUND").get();

assertThat(myProp, equalTo("myprop_value_default"));

Archaius can load property appropriate to an environment, consider that there is a "sample-perf.properties" with the same configuration over-ridden for perf environment:

myprop=myprop_value_perf

Now Archaius can be instructed to load the configuration in a cascaded way by adding the following in sample.properties file:

myprop=myprop_value_default

@next=sample-${@environment}.properties

And the test would look like this:

ConfigurationManager.getDeploymentContext().setDeploymentEnvironment("perf");
ConfigurationManager
        .loadCascadedPropertiesFromResources("sample");

String myProp = DynamicPropertyFactory.getInstance().getStringProperty("myprop", "NOT FOUND").get();

assertThat(myProp, equalTo("myprop_value_perf"));

Spring Property basics

Spring property basics are very well explained at the Spring Framework reference site here. In short, if there is a property file "sample.properties", it can be loaded up and referenced the following way:

@Configuration
@PropertySource("classpath:/sample.properties")
public class AppConfig {
    @Autowired
    Environment env;

    @Bean
    public TestBean testBean() {
        TestBean testBean = new TestBean();
        testBean.setName(env.getProperty("myprop"));
        return testBean;
    }


}

Or even simpler, they can be de-referenced with placeholders this way:

@Configuration
@PropertySource("classpath:/sample.properties")
public class AppConfig {
    @Value("${myprop}")
    private String myProp;

    @Bean
    public TestBean testBean() {
        TestBean testBean = new TestBean();
        testBean.setName(myProp));
        return testBean;
    }

    @Bean
    public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
        return new PropertySourcesPlaceholderConfigurer();
    }

}

Making Archaius properties visible to Spring

So now the question is how to get the Archaius properties visible in Spring, the approach I have taken is a little quick and dirty one but can be cleaned up to suite your needs. My approach is to define a Spring PropertySource which internally delegates to Archaius:

import com.netflix.config.ConfigurationManager;
import com.netflix.config.DynamicPropertyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.PropertySource;

import java.io.IOException;

public class SpringArchaiusPropertySource extends PropertySource<Void> {


    private static final Logger LOGGER = LoggerFactory.getLogger(SpringArchaiusPropertySource.class);


    public SpringArchaiusPropertySource(String name) {
        super(name);
        try {
            ConfigurationManager
                    .loadCascadedPropertiesFromResources(name);
        } catch (IOException e) {
            LOGGER.warn(
                    "Cannot find the properties specified : {}", name);
        }

    }

    @Override
    public Object getProperty(String name) {
         return DynamicPropertyFactory.getInstance().getStringProperty(name, null).get();
    }
}

The tricky part is registering this new PropertySource with Spring, this can be done using an ApplicationContextInitializer which is triggered before the application context is initialized:

import com.netflix.config.ConfigurationBasedDeploymentContext;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.util.StringUtils;

public class SpringProfileSettingApplicationContextInitializer
        implements ApplicationContextInitializer<ConfigurableApplicationContext> {

    @Override
    public void initialize(ConfigurableApplicationContext ctx) {
        ctx.getEnvironment()
                .getPropertySources()
                .addFirst(new SpringArchaiusPropertySource("samples"));
    }
}

And finally registering this new ApplicationContextInitializer with Spring is described here

This is essentially it, now the Netflix Archaius properties should work in a Spring application.

Java 8 Stream to Rx-Java Observable

I was recently looking at a way to convert a Java 8 Stream to Rx-Java Observable.

There is one api in Observable that appears to do this :

public static final <T> Observable<T> from(java.lang.Iterable<? extends T> iterable)

So now the question is how do we transform a Stream to an Iterable. Stream does not implement the Iterable interface, and there are good reasons for this. So to return an Iterable from a Stream, you can do the following:

Iterable iterable = new Iterable() {
    @Override
    public Iterator iterator() {
        return aStream.iterator();
    }
};

Observable.from(iterable);

Since Iterable is a Java 8 functional interface, this can be simplified to the following using Java 8 Lambda expressions!:

Observable.from(aStream::iterator);

First look it does appear cryptic, however if it is seen as a way to simplify the expanded form of Iterable then it slowly starts to make sense.

Reference:
This is entirely based on what I read on this Stackoverflow question.