IMHO the core idea of using a Flow Publisher / Subscriber is to have more than one item by definition. Otherwise one would simply use Provider / Consumer. Using Flow to process single instances feels like using Collection instead of singletons.
-Markus
From: Pavel Bucek [mailto:pavel.bucek_at_oracle.com]
Sent: Freitag, 17. März 2017 08:46
To: jsr370-experts_at_jax-rs-spec.java.net
Subject: NIO API review / Publisher<Pojo> vs Mono<Pojo>
Dear EG members,
thanks for the feedback provided for the initial topic.
I do see it that there is a general agreement that using Flow APIs is a way we'd like to explore.
One of the part of problems we are trying to solve is how to provide an access to the entity when mapped to POJOs. Source<POJO> can produce infinite list of POJOs, but http entities are generally finite (let's not include SSE yet). Do you have any thoughts about having an ability to signal that the resource method will consume/produce only single pojo instance?
@POST
@Path("/ex2")
@Consumes(MediaType.APPLICATION_JSON)
public void ex2(Flow.Publisher<POJO> entity,
@Suspended AsyncResponse response) {
// ...
}
Consider snippet above.
JAX-RS implemenantions most likely won't handle separators, since they are custom and we are not going to specify the format of the entity in any way, it's up to the MessageBodyReaders/Writers. The implementation will most likely provide only support for serializing/deserializing single pojo, as it does now.
There is relevant comment (linked also from the kickoff email):
// Multiple items vs single one
//
// The #writeTo method can be written in a way, which can react to a stream or single event and producing
// appropriate output. Something like:
//
// entityPojoPublisher.subscribe(new Flow.Subscriber<NioResource.POJO>() {
//
// volatile NioResource.POJO cache = null;
// volatile boolean moreThanOne = false;
//
// @Override
// public void onSubscribe(Flow.Subscription subscription) {
// subscription.request(Long.MAX_VALUE);
// }
//
// @Override
// public void onNext(NioResource.POJO item) {
// if (item == null) {
// this.cache = item;
// } else {
// if (moreThanOne) {
// entity.onNext(/* separator if needed */ null);
// entity.onNext(/* pojo2bytes(item) */ null);
// } else {
// moreThanOne = true;
//
// entity.onNext(/* root elem start */ null);
// entity.onNext(/* pojo2bytes(cache) */ null);
// entity.onNext(/* separator if needed */ null);
// entity.onNext(/* pojo2bytes(item) */ null);
// }
// }
// }
//
// @Override
// public void onError(Throwable throwable) {
// // handle Error
// }
//
// @Override
// public void onComplete() {
// if (!moreThanOne) {
// entity.onNext(/* pojo2bytes(cache) */ null);
// } else {
// entity.onNext(/* root elem end */ null);
// }
// entity.onComplete();
// }
// });
This is obviously valid for both client and server side and for consuming and producing. Producing might be little different (since we CAT return stream of POJOs (we do have SSE), so let's focus on server and consuming part.
As stated in in the subject, do we need something like "Mono" (publisher guaranteed to publish only single event)?
Looking forward for any comments.
Thanks and regards,
Pavel