Forem: Gadi Eichhorn

Surprising Qualities of Event Driven System

Gadi Eichhorn — Sat, 10 Jul 2021 07:12:08 +0000

Few discoveries I made during the last two years working on a small java project (that grow big) for a client using Java 8, Vertx.io, RxJava and OSGi.

I consider these new (to me) as I don't recall reading about them anywhere else. Comments are welcomed.

Fail and Retry

Normally when exception happens you need to address the exception by different means, but what if you want to try again?

Let's imagine we are calling a remote system or waiting for a network device to reconnect, the most sensible process will be to retry the operation, now on a normal call a service process you will have to develop some loop of try catch to keep trying the call until successful, this creates ugly code that appears in many places in the codebase and hard to test and refactor.

If the process was triggered with an event (assuming you have some bus that delivered this event to you) you can apply a retry strategy by simply wrap the event in a retry event with information about the time to wait or the number of attempts and publish the event again. The retry event handler job is just to handle the wait/sleep and trigger the inner event again (and again). Your business login only needs to give information about the retry strategy (delay/attempts) and nothing more. All the login is handled in one place which is the event retry strategy handler.

Never leak API specification

Designing API is hard (it is really hard) because any change you make to your API you have to refactor all the consumers (and tests) and they may have cascading effects for the change in unexpected places.

How about events? if you are an event subscriber you only interested in consuming the event and you are not publishing any API as no one is calling you. This means you don't have an API to manage.

You can still provide output by using the event bus and not to any direct callers.

Parallel features and refactoring

Imagine there is a change required to the system… like that never happens.

Well, with event-driven design you can simply develop your new feature as a separate parallel effort, there is no need to change the old code, once the feature/fix is in place we can simply disable the old process (handler) and the system will not know of the change (this is simplified of course as you need to make sure all the work the old code was doing is not needed anymore or replaced with the new code).

The best example I know of is CQRS, where there can be many subscribers to an event, each implementing a different process handling the information delivered, the term coined is projections.

In principle, the old and new event handlers can happily live together in the system (in case they don't override each other work), this eliminates the need to refactor the old handler to be the new handler. Switch one ON and switch the other OFF.

Thanks for reading.

Rxjava Message Bus for OSGi

Gadi Eichhorn — Sat, 10 Jul 2021 07:04:19 +0000

I wanted to make OSGi components communicate in a rich event driven way where I can publish many different messages between components without tight coupling or direct @Reference(ing). The OSGI EventAdmin is a nice way to start but sooner or later you run into limitations.

The drive was to make a system that is more Reactive and easy to understand but at the same time not limiting.

I chose to use RxJava as the underlying engine for delivering messages because:

It is fast (async)
It is simple
It is powerful with operations & schedulers
it has great unit test tools build in.

I did have few challenges on the way, mainly making a very clean API that will be easy to share and ruse and not complicated to understand. I also needed a good way to unit test my components and during OSGi integration tests.

As we will see some tests are a bit more tricky then others.

I got inspiration from this blog which I recommend reading as I am not going to repeat what Paulina described. I decided to port the implementation to OSGi and create a small configuration abstraction.

https://medium.com/@PaulinaSadowska/writing-unit-tests-on-asynchronous-events-with-rxjava-and-rxkotlin-1616a27f69aa

A simple bus implementation

A very simple message bus with a generic Message type as a payload. Clients are required to implement messages of this type:

import com.rds.demo.api.Message;

public class MyMessage implements Message {
}

Now we need to get instance of the MessageBus so we can start communicating with other components using RxJava
The Message bus allow two main methods to consume and produce messages on the bus.

public interface MessageBus<T> {

    void publish(T message);

    void subscribe(Observer<T> observer);

}

The default MessageBusProvider then implement this service and deploy it as OSGi service. Internally using the PublishSubject that allows dynamically sending messages and subscribing to them as a hot Observable.

Pay attention to the SINGLETON annotation making sure there is only one Bus instance in the OSGi registration.

@Component(service = MessageBus.class, scope = ServiceScope.SINGLETON)
public class MessageBusProvider<T extends Message> implements MessageBus<T> {

    private final PublishSubject<T> publishSubject = PublishSubject.create();

    @Override
    public void publish(T message) {
        publishSubject.onNext(message);
    }

    @Override
    public void subscribe(Observer<T> observer) {
        publishSubject.subscribe(observer);
    }

}

Now if we want to test the new service directly (unit test not OSGi integration tests) we can use the RxJava test abstractions like so:


@Test
public void publishMessageTest() {
    TestObserver<Message> test = TestObserver.create();

    MessageBus<Message> bus = new MessageBusProvider<>();

    bus.subscribe(test);
    bus.publish(new TestMessage());

    test.assertValueCount(1);
    test.assertNoErrors();
    test.assertNotComplete();
}

Step two, Schedulers

That was easy, right? Well not so fast... we forgot Schedulers. RxJava Schedulers allow running pipes on different threads. This is the real power of RxJava when it comes to doing things in parallel.

So lets add a computation scheduler to the mix:

    @Override
    public void subscribe(Observer<T> observer) {
        publishSubject
        .subscribeOn(schedulers.computation())
        .subscribe(observer);
    }

But now we have a problem as our Observers are listening on the computation thread pool our testObserver
will not work anymore as it will complete without waiting.

To overcome this problem we need to use a different Scheduler inside our tests but our MessageBus service is not allowing us to change anything so we have a problem.

To extends the MessageBus one option is to allow injection of another service that will control the Schedulers. As Paulina outlined, this can be an abstraction of the Schedulers as a service. SIne we are not concern with Android UI I made some changes to the API.

public interface SchedulersFactory {

    Scheduler blocking();

    Scheduler pooled();

}

Injecting this service into our MessageBus service we can now change the code to be:

@Component(service = MessageBus.class, scope = ServiceScope.SINGLETON)
public class MessageBusProvider<T extends Message> implements MessageBus<T> {

    private final PublishSubject<T> publishSubject = PublishSubject.create();

    @Reference
    private SchedulersFactory schedulersFactory;

    public MessageBusProvider() {
    }

    public MessageBusProvider(SchedulersFactory schedulersFactory) {
        this.schedulersFactory = schedulersFactory;
    }

    @Override
    public void publish(T message) {
        publishSubject.onNext(message);
    }

    @Override
    public void subscribe(Observer<T> observer) {
        publishSubject.subscribeOn(schedulersFactory.pooled()).subscribe(observer);
    }

}

The default constructor is needed for the OSGi framework to create instance of our class.
The Other constructor will allow us to inject a different SchedulerFactory in our tests.

public class TestSchedulersFactory implements SchedulersFactory {

    @Override
    public Scheduler pooled() {
        return Schedulers.trampoline();
    }

    @Override
    public Scheduler blocking() {
        return Schedulers.trampoline();
    }

}

The trampoline Scheduler will fix our problem as it perform operations in sequence so our test observer will be tested only after all the messages were processed.

Now we can change the using tests to inject a different SchedulersFactory instance.

       MessageBus<Message> bus = new MessageBusProvider<>(new TestSchedulersFactory());

All good for now.

Integration tests

OSGi testing is considered integration tests as it attempt to build a full runtime and inject unit test wrapper into the runtime, then run assertions. This allows us to build small parts of our application, adapt the configuration, and test everything is wired properly. But are we forgetting something? Yes, test will still fail as we didn't plan for SchedulerFactory configuration.

Let's fix that.

First we need to allow configuration on our SchedulersFactory, our component will now have a configuration interface

public @interface SchedulersFactoryProviderConfiguration {

    @AttributeDefinition(
            name = ".blocking",
            type = AttributeType.STRING,
            description = "Blocking scheduler",
            required = false,
            options = {
                    @Option(label = "io", value = "IO"),
                    @Option(label = "computation", value = "COMPUTATION")
            }
    )
    String _blocking() default "IO";

    @AttributeDefinition(
            name = ".pooled",
            type = AttributeType.STRING,
            description = "Pooled scheduler",
            required = false,
            options = {
                    @Option(label = "io", value = "IO"),
                    @Option(label = "computation", value = "COMPUTATION")
            }
    )
    String _pooled() default "COMPUTATION";

}

We can then inject this configuration to our component in the activate method

@Component(
        name = "com.rds.reactive.provider.schedulers.provider",
        service = SchedulersFactory.class,
        configurationPolicy = ConfigurationPolicy.OPTIONAL)
@Designate(
        ocd = SchedulersFactoryProviderConfiguration.class
)
public class SchedulersFactoryProvider implements SchedulersFactory {

    private Scheduler pooled;
    private Scheduler blocking;

    @Activate
    public void activate(SchedulersFactoryProviderConfiguration cfg) {
        System.out.println("ACTIVATE");
        blocking = SchedulerType.get(cfg._blocking()).value();
        pooled = SchedulerType.get(cfg._pooled()).value();
    }

    @Override
    public Scheduler blocking() {
        return blocking;
    }

    @Override
    public Scheduler pooled() {
        return pooled;
    }

}

And with a small helper enum we are making sure we have the right Schedulers defined.

public enum SchedulerType {

    IO(Schedulers.io()),
    COMPUTATION(Schedulers.computation()),
    TRAMPOLINE(Schedulers.trampoline());

    private final Scheduler scheduler;

    SchedulerType(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    Scheduler value() {
        return scheduler;
    }

    private static final Map<String, SchedulerType> ENUM_MAP;

    static {
        Map<String, SchedulerType> map = new HashMap<>();
        for (SchedulerType instance : SchedulerType.values()) {
            map.put(instance.name(), instance);
        }
        ENUM_MAP = Collections.unmodifiableMap(map);
    }

    public static SchedulerType get(String name) {
        return ENUM_MAP.getOrDefault(name, COMPUTATION);
    }
}

Our SchedulerFactory is injected into the MessageBus using a @Reference (OSGi DS).

    @Reference
    private SchedulersFactory schedulersFactory;

One last thing, we need to tell OSGi framework how to configure our SchedulersFactory in test mode as we still need to use the trampoline scheduler in our tests.

In our config folder (see rest.runbnd) we link the config folder to our runtime:

-runproperties: \
    felix.fileinstall.dir=${.}/config

Add the configuration properties file to this folder:

config/com.rds.reactive.provider.schedulers.provider.cfg

.blocking=TRAMPOILNE
.pooled=TRAMPOLINE

And now for our integration tests... we need to get an instance of the MessageBus from the framework, this instance will use the SchedulersFactory instance with our TRAMPOLINE override configuration so we can have a predictable working tests.

public class MessageBusTest {

    private final BundleContext context = FrameworkUtil.getBundle(MessageBusTest.class).getBundleContext();

    private ServiceTracker<MessageBus, MessageBus> messageBusMessageBusServiceTracker;

    private MessageBus bus;

    @Before
    public void before() throws InterruptedException {
        messageBusMessageBusServiceTracker = new ServiceTracker<MessageBus, MessageBus>(context, MessageBus.class, null);
        messageBusMessageBusServiceTracker.open();
        bus = messageBusMessageBusServiceTracker.waitForService(500);
        Assert.assertNotNull(bus);
    }

    @After
    public void after() {
        messageBusMessageBusServiceTracker.close();
    }

    @Test
    public void canPublishToBusTest() {
        TestObserver<Message> test = TestObserver.create();
        bus.subscribe(test);
        bus.publish(new TestMessage());
        test.assertValueCount(1);
        test.assertNotComplete();
        test.assertNoErrors();
    }

    @Test
    public void multipleMessagesTest() {
        TestObserver<Message> test = TestObserver.create();
        bus.subscribe(test);

        // it is critical to set the scheduler here otherwise it is using the inmternal computation scheduler.
        Observable.interval(10, TimeUnit.MILLISECONDS, Schedulers.trampoline())
                .take(10)
                .map(t -> new TestMessage())
                .subscribe(testMessage -> bus.publish(testMessage));

        test.assertValueCount(10);
        test.assertNotComplete();
        test.assertNoErrors();
    }

}

Notice we ask for the MessageBus from the context and that is it. The service was configured for us using the cfg file.

Also notice the interval Observer is using a scheduler. If you not giving the interval a scheduler it will default to computation and break our tests.

        Observable.interval(10, TimeUnit.MILLISECONDS, Schedulers.trampoline())

Summary

RxJAva gives us powerful tools for building and testing applications but we need to be careful when testing them.

I hope this was helpful if you are working with OSGI or RxJava.

The code

Please feel free to use, port and contribute.

https://github.com/gadieichhorn/osgi-reactive

References

Paulina's blog post -> https://medium.com/@PaulinaSadowska/writing-unit-tests-on-asynchronous-events-with-rxjava-and-rxkotlin-1616a27f69aa

EffectiveOSGi -> https://github.com/effectiveosgi

Pass the logger

Gadi Eichhorn — Sat, 10 Jul 2021 06:48:39 +0000

I call this the Logger design pattern :)

We are lucky to have a Logger API (slf4J) which allows us to choose implementation as we see fit, I specifically like loggback as it offers simple to advance configuration.

One trick I use when dealing with class abstraction and inheritance is logging using the concrete class and not using any parent logger.

I found two main ways to pass the logger to make sure I log using the right Logger instance.

Loggers are static

public abstract class Parent  {

    private static final Logger log = LoggerFactory.getLogger(Parent.class);

    protected void doSomething() {
        // will log using this class Logger
        log.info("Success");
    }
}

My typical log message will be:

04-07-2019T11:00:00.662 [main] INFO Parent.doSomething Success

We have some default methods we would like to use in our inherited classes. But now any inherited class will log using the Parent class logger.

public class Child extends Parent {

    private static final Logger log = LoggerFactory.getLogger(Child.class);

    protected void doSomethingElse() {
        // will log using this class Logger
        log.info("Some message");
        doSomething(); // will log using parent logger
    }

}

Now the log is coming from two different classes using two different Loggers

04-05-2019T11:11:00.662 [main] INFO Child.doSomethingElse Some Message
04-05-2019T11:11:00.662 [main] INFO Parent.doSomething Success

It will be useful to have both log messages with the same Child class message as we might have many derived classes and we need to know the origin caller of doSomething()

Constructor override

We can allow the derived classes to pass the Logger into the constructor:

public abstract class Parent  {

    private final Logger logger;

    public Parent(Logger logger) {
        this.logger = logger;
    }

    protected void doSomething() {
        // will log using the injected Logger instance
        log.info("Success");
    }

}

Our Child class will have to pass the static instance of the Logger to our Parent class

public class Child extends Parent {

    private static final Logger log = LoggerFactory.getLogger(Child.class);

    public Child() {
        super(log);
    }

    protected void doSomethingElse() {
        // All logging will be done using this instance Logger
        log.info("Some message");
        doSomething();
    }

}

Log will now have the Child class name in the log.

04-05-2019T11:21:00.332 [main] INFO Child.doSomethingElse Some Message
04-05-2019T11:21:00.332 [main] INFO Child.doSomething Success

Method override

In this case we might want to mix the two loggers where some logging will be done using the parent logger and some will be done using our child class. We can pass the Logger instance to the parent method call.

public abstract class Parent  {

    protected void doSomething(Logger log) {
        // will log using the injected Logger instance
        log.info("Success");
    }

}

And in the child class we don't need the constructor anymore and we simply pass the Logger to the method.

public class Child extends Parent {

    private static final Logger log = LoggerFactory.getLogger(Child.class);

    protected void doSomethingElse() {
        // All logging will be done using this instance Logger
        log.info("Some message");
        doSomething(log); // we pass the logger to the method
    }

}

Log will now have the Child class name in the log.

04-05-2019 11:31:00.100 [main] INFO Child:doSomethingElse Some Message
04-05-2019 11:31:00.100 [main] INFO Child:doSomething Success

Mix and match

There are many ways to go about implementing these ideas with default Logger and Pass the logger so I was just outlining the edge cases. You can mix and match the two ideas in many different ways.

If you find other ideas useful please share in the comments.

Hope you find this short post helpful.

Thanks for reading.