| [jnosql-dev] Create a Reactive Stream as an extension |
The goal of this proposal is to write an extension to each API that allows integration with Reactive Streams API.
Briefly, Reactive Stream is an initiative to provide a standard for asynchronous stream processing with non-blocking backpressure.
This extension will work integrated with the Eclipse MicroProfile Reactive Streams Operators.
Reactive Streams is an integration SPI - it allows two different libraries that provide asynchronous streaming to be able to stream data to and from each other.
Reactive Streams is not however designed to be used directly by application developers. The semantics defined by Reactive Streams are very strict, and are nontrivial, particularly in the areas of thread safety, to implement correctly. Typically, application developers are expected to use third-party libraries that provide the tools necessary to manipulate and control streams. Examples include Akka Streams, RxJava, and Reactor.
The repositories programming model is the most high-level abstraction and we’ll have a repository to handle with ReactiveRepository:
public interface GodRepository extends ReactiveRepository<God, String> {
}
String id =...;
Publisher<God> publisher = repository.findById("id");
CompletionSubscriber<God, Optional<God>> subscriber = ReactiveStreams.<God>builder().findFirst().build();
publisher.subscribe(subscriber);
CompletionStage<Optional<God>> completion = subscriber.getCompletion();
Just as the traditional repositories are based on traditional template implementations, the reactive ones are built on top of a reactive template.
ReactiveKeyValueTemplate template
Publisher<God> publisher = template.get("id", God.class);
CompletionSubscriber<God, Optional<God>> subscriber = ReactiveStreams.<God>builder().findFirst().build();
publisher.subscribe(subscriber);
CompletionStage<Optional<God>> completion = subscriber.getCompletion();
References: