Hi all, I wanted to start a discussion about the use of Flow.Subscriber and Flow.Publisher in JEP 321 (HTTP Client API).
It seems that users are required to implement their own publishers and subscribers, that is, they can't take a Flow.Publisher or Flow.Subscriber provided by another reactive streams implementation, and pass it on to the HttpClient API. The reason for this is that the HttpClient API doesn't accept Flow.Publisher/Flow.Subscriber, rather it extends them in HttpRequest.BodyPublisher and HttpResponse.BodySubscriber, and then requires the user to return instances of those sub interfaces from their BodyHandlers. Let's say I have a database driver that produces a Flow.Subscriber for consuming a stream to be stored in the database, and I want to plumb a response body into that database subscriber. I can't return this from an HttpResponse.BodyHandler it requires me to return a HttpResponse.BodySubscriber, not a Flow.Subscriber. Of course, users can implement their own HttpResponse.BodySubscriber that delegates onSubscribe/onNext/onComplete/onError to the databases Flow.Subscriber, but needing to do this every time does not provide a great developer experience. In order for reactive streams implementations to integrate with each other, Flow.Subscriber and Flow.Publisher should only be extended if the implementation is providing its own implementation of that interface and doesn't expect end users to implement it. For cases where an implementation expects users either to pass that sub interface, or return it from a method to that they implement as is the case for BodyHandler, then it has to be a Flow.Subscriber or Flow.Publisher, not a sub interface. So perhaps HttpResponse.BodySubscriber should be modified to, rather than extending Flow.Subscriber, have a getSubscriber() method that returns a Flow.Subscriber. Alternatively, the API could be inverted, for example, HttpResponse.BodyHandler.apply could be modified to take three parameters, the status code, the HttpHeaders, and a Flow.Publisher, and the return value could be CompletionStage<T>. This approach would actually maximise interoperability, since a lot of reactive streams implementations are publisher centric. For example, to my knowledge RxJava only provides limited support for creating and transforming data using a Flow.Subscriber, whereas if you give RxJava a Flow.Publisher, you can then do the full range of operations (eg map) that RxJava supports on that stream (someone with a better understanding of RxJava correct me if I'm wrong). There's probably many other solutions, these are just two that I've thought of. There is also a broader discussion that needs to be had in the reactive streams community over whether there should be a bias towards everything returning/accepting publishers, or if implementations should evenly support both. Supporting one or the other arbitrarily will have interoperability implications, and while it's definitely beyond the scope of JEP 321 to decide on that, it's something that we should keep in mind. Regards, James -- *James Roper* *Senior Octonaut* Lightbend <https://www.lightbend.com/> – Build reactive apps! Twitter: @jroper <https://twitter.com/jroper>
