futura - building blocks for asynchronous programming in Clojure

It there several different ways to create a promise in futura library. You can create it inmediatelly resolved with initial value, inmediatelly rejected with some exception.

Let start with a basic example using the commonly known promise delivering in clojure:

An other way to create a promise is using a factory function that can resolve or reject promise in asynchrnous way. If you are familiar with javascript promises, you will found that very similar:

The futura's promises can be used as drop in replacement for clojure promises, because them offers also blocking operations:

If you try to deref a promise that is rejected, the exception will be rereaised in the calling thread.

futura provides useful predicates that will allow check the state of a promise in any time.

Let see some examples:

The done? predicate checks if a promise is fullfiled, independently if is resolved or rejected.

Futura also offers powerful chaining methods for allow easy compose async computations. In previous examples we have seen then function, let see an other more complex example using it:

It also exposes a chain method for error handling:

The catch chain function also return a promise. Promise that will be resolved or rejected depending on that will happen inside the catch handler.

In some circumstances you will want wait a completion of few promises at same time, and futura also provides helpers for that:

The futura's streams api is really very simple. It consists in ver few functions.

Let’s go to create our first publisher:

The publisher function without additional parameters creates a empty, and unbuffered publisher. Now you can put the items to the using the put! function:

The return value of the put! function is a promise that will be resolved to true when the value is accepted by the publisher. And in case of the unbuffered publisher it will happens when one subscription will request a value.

And finally, for obtain values from the publisher, you should create a subscription:

The return value of take! function is also a promise and will be resolved with a value when the first one will be available.

Take care that the subscribe function it is not analogous to the .subscribe method of the publisher. Instead of creating opaque object that acts as relation betwen the publisher and the subscriber, the subscribe function creates a open object that does not has direct relation with any subscriber.

Obviously, behind the schenes the subscribe function uses the publishers .subscribe method. So the publisher implementation is completelly interopreable with other third party libraries and the java world.

This subscription strategy has some advantages over the purposed by the reactive-streams, because you can treat it like a subscription stream. It implements convenient interfaces for the ability to treat the subscription like a clojure sequence, java iterable or even core.async channel.

You should known that everything in this implementation is lazy. Creating subscription does not automatically request all items from publisher. A subscription will only request a item to the publisher when you are request it.

Behind the scenes, publisher and subscription are implemented using core.async, so the nil values are does not allowed and represent that the publisher or the subscription is closed. The subscription as you can observe implements the .close() method that allows use it with with-open macro, but take care, the publisher can end early and subscrion will be closed also before with-open calls the .close() method.

As we said previously, the real purpose of this abstraction is using it as connection with java world, surelly if you are clojure only user you may want use core.async or manifold directly without additional layer.

In fact, futura is actually used by catacumba for comunicating asyncronous streams with ratpack/netty. This abstraction is choiced because it has support for backpressure that is very important in asynchronous network applications.

But for use it as a intermediary layer with third party libraries you should will be able create publisher from the existing abstractions in clojure. Now, let se how you can do it:

As you can observe the previous example, you can see that the publisher can be easily converted to sequence. That happens because publisher implements the clojure’s Seqable interface that behind the scenes uses a subscription and blocking access to all items until the subscription is closed.

So, you can create a publisher from this kind of abstractions: promise, jdk8 completable future, manifold deferred, manifold stream, sequences and iterables.

Additionally to the previously mentioned abstractions, you can create a publisher from another publisher using the transform function. This kind of composition allows you attach a transducer for apply some kind of transformations to the stream values.

Take care de some implementation detail: the transducer is applied for subscriptions, not for the publisher. So, if you use some kind of (take N) transducer with possible infinite publisher, it will cause that each subscription will be closed after N items but the publisher will remain active accepting new subscriptions.

One interesting thing is that open subscriptions can be used as channels so them are fully compatible for usage in core.async go block like any other channel:

The atomic reference privides a lock-free, thread safe object reference container. The real purpose of this type is store a reference to an arbitrary object and will be able mutate it in a thread safe, lock-fre way.

This is a specialized version of Atomic Reference that is higtly optimized for boolean values. It has the same operations and implements the same abstractions that previously explained atomic ref.

You can create an atomic boolean using boolean function from futura.atomic namespace:

This is a specialized version of Atomic Reference that is highly optimized for numeric operations with longs. It has the same operations and implements the same abstractions that previously explained ref and boolean atomic types.

But additionally implements a complementary abstraction that brings some powerfull operations that only fits for numeric types, such as atomic counters and similars.

You can create an atomic long using long function from futura.atomic namespace:

Here some examples using the functions defined for numeric atomic types, such as "get and increment/decrement" operations:

And optionally you can increment it with user specified delta:

Both libraries offers similar abstractions, futura offers promises and streams and manifold offers deferreds and streams. The main difference of this libraries is clearly philosoficaly:

The futura library obviosly is less mature that manifold because of obvios reasons that manifold exists some time ago.