Re: websockets

[Konrad Malawski]

Hello everyone,
seems I was called out by name so wanted to confirm that point is being
understood correctly.
(tracking this thread with much anticipation, but don’t want to add more
noise, discussion seems pretty healthy so far!).

Rossen clarifies my point very well, but to confirm it myself as well since
I was quoted by name on it:

The argument about not extending RS/Flowable interfaces directly matters
more for libraries that are outside of the JDK (before j.u.c.Flow existed),
such as Akka Streams or Reactor, since they were adopting the interfaces
early and showing that those things do work,
and that there is a will and ecosystem ready to adopt them.

In the JDK of course the decision is simple — one simply shall use the
types available in the JDK: java.util.concurrent.Flowable,
and ignore that the org.reactivestreams.* types exist at all. The contract
is the same indeed, so the RS ecosystem now has to work
well with both for a while, but anything that is being built anew (esp in
the JDK), can assume Flow is the right thing to use.

I hope we’d be able to figure out a way to make a Flow-compatible API
available, in order to grow the number of connectable APIs by this
interface,
same as it would be lovely to have File access types implement them as well.

I’m a bit lost what the current main question to answer (or prototype) here
is, that would be nice to focus on, to decide the viability of such API?

-- 
Cheers,
Konrad 'ktoso ' Malawski
Akka  @ Lightbend 

On February 22, 2018 at 9:04:22, Rossen Stoyanchev (rstoyanc...@pivotal.io)
wrote:

​​hi,

> Simone Bordet:
> As the WebSocket APIs were moved to incubation and are now slated
> for JDK 11, it may be worth considering again whether it's the
> case to provide a Flow-based APIs.

I second that. There were two main issues pointed out in this thread:

> The first one is how to communicate errors back to the user's
> code that publishes messages to WebSocket.

Isn't it mostly low level, transport errors about which the application
can't do much about, but clean up and close the connection? Arguably a
cancellation is adequate, or do you have more specific errors and cases in
mind? Alternatively, the JSR-356 WebSocket API provides a single onError
callback on the Endpoint type. I haven't come across any limitations around
that.

> The second issue is how to communicate completion signals from
> processing individual messages.

James Roper​ mentioned this already, but the Netty PooledByteBufAllocator
demonstrates it's feasible to recycle buffers transparently without the
need for a per-message confirmation. I don't think the argument holds that
not having per message completion signals forces unnecessary allocations,
garbage creation and copying.

There were a "dozen of smaller issues" mentioned. However there were no
links to earlier discussions, so I'm not sure what those were.

The main purpose of Reactive Streams is to connect components from
different libraries. The currently proposed WebSocket API cannot be easily
integrated into a Reactive Streams pipeline. One would have to build a
bridge to Reactive Streams first and I don't think that should be left as a
separate exercise, since Flow is in the JDK and it's the reason why it was
introduced, to provide a common contract for such scenarios.

> Another point that makes me feeling uncertain is the good comment by
> Konrad Malawski about the design of Akka Streams versus e.g. Spring's
> Flowable, where the former does not implement RS or Flow interfaces,
> while the latter does.

One minor clarification first. By "Spring's Flowable" I presume you mean
the Flux and Mono types from the Reactor project? Reactor does not depend
on any Spring projects, and it is a lower level project. It's directly
comparable to RxJava, in fact it implements the same ReactiveX patterns,
but more explicitly targeting server-side Java applications vs Android.

To the point, I don't think the Flow vs RS contracts should a cause for
hesitation. HttpClient already uses the Flow contract and that's the
obvious choice for WebSocket too. More broadly, the JDK Flow and the RS
contracts are identical, and have identical semantics, so adapting from one
to the other is trivial. In most cases applications are not consuming these
contracts directly, which is also Konrad's point, so practically speaking I
don't see much dichotomy there.​

Regards,
Rossen

​

Re: websockets

[Chuck Davis]

Well, yes, WebSocket on the server is an entirely different beast (at this
time -- although perhaps jdk9 WebSocket will make it's way to JEE someday)
-- and that definitely has to work with browsers.  But the whole world is
not going to browsers as much as some would have us believe.  The desktop
is alive and well for internal users at businesses.  And the quality of
most of that software is so incredibly dismal one wonders how any of it
gets sold.  JavaFX and WebSockets can put the FIX on that stuff besides
being cross-platform with update repositories!  Long live desktop
applications and long live the servers that power them!! 邏  Oh, yeah, and
long live WebSockets!!


On Sun, Mar 4, 2018 at 4:24 PM, James Roper <ja...@lightbend.com> wrote:

> Hi Chuck,
>
> I'm definitely not saying that the Java client is for browsers. I'm saying
> that the WebSocket protocol was designed to be used by the API provided by
> web browsers. No protocol is ever implemented on an island, it is
> implemented with an intended use case. To best understand a protocol, to
> understand why it is the way it is, to understand what would be a good way
> to use it, and what might be a bad way to use it, and therefore to
> understand what an API for it should look like, it is good to understand
> what use case the protocol was designed for. And that use case was to be
> used from browsers. So when you create a WebSocket client, or server, in
> any language, it is very valuable to look at the browser provided APIs to
> understand how the protocol was intended to be used, and therefore what
> features you should expose to users.
>
> To put in perspective, most server implementations of WebSockets are
> designed to be connected to from browsers, non browser clients are still a
> relatively niche use case for WebSockets. And so servers are going to look
> to the browser for what features that should give priority to. If a browser
> doesn't support X, then it's likely that a server implemention either won't
> support X, or won't provide very good or easy to use support for X.
> Consequently, it's of dubious value to provide a new client where support
> for X complicates the API, since servers aren't going to support that well
> anyway, it's providing a feature that you won't be able to use at the cost
> of making other features harder to use.
>
> Regards,
>
> James
>
>

Re: websockets

[James Roper]

Hi Chuck,

I'm definitely not saying that the Java client is for browsers. I'm saying
that the WebSocket protocol was designed to be used by the API provided by
web browsers. No protocol is ever implemented on an island, it is
implemented with an intended use case. To best understand a protocol, to
understand why it is the way it is, to understand what would be a good way
to use it, and what might be a bad way to use it, and therefore to
understand what an API for it should look like, it is good to understand
what use case the protocol was designed for. And that use case was to be
used from browsers. So when you create a WebSocket client, or server, in
any language, it is very valuable to look at the browser provided APIs to
understand how the protocol was intended to be used, and therefore what
features you should expose to users.

To put in perspective, most server implementations of WebSockets are
designed to be connected to from browsers, non browser clients are still a
relatively niche use case for WebSockets. And so servers are going to look
to the browser for what features that should give priority to. If a browser
doesn't support X, then it's likely that a server implemention either won't
support X, or won't provide very good or easy to use support for X.
Consequently, it's of dubious value to provide a new client where support
for X complicates the API, since servers aren't going to support that well
anyway, it's providing a feature that you won't be able to use at the cost
of making other features harder to use.

Regards,

James

On 4 March 2018 at 05:49, Chuck Davis <cjgun...@gmail.com> wrote:

> James, this does not make any sense to me.  Last I heard browsers no
> longer run java.  Why would JSE give a hang about designing a
> communications mechanism for browsers on the client side?  That is the job
> of the browser developers.  On  the other hand, JSE WebSocket is the best
> thing to come along for java clients since RMI which was released I think
> in 1.1.  Java WebSocket is NOT for browsersit's for Java clients
>   And I'm quite pleased with the progress I've made refactoring to use
> the jdk9 WebSocket api together with wildfly11 on the server (and still
> blazingly FAST -- seems to be snappier than jdk8 implementation).  Now if
> the IDEs would get in gear to fully implement jdk9 I'd be in fat city.
>
> And my thanks to everyone who has participated in this particular
> conversation on this list and all the expertise that has been shared.
>
> >Yes, there's no formal definition of high/low level, but in the case of
> WebSockets, we can draw some lines based roughly on what web browsers
> (which is the target application developer experience that >WebSockets were
> designed for) support as being high level, and what the WebSocket protocol
> allows beyond this as low level.
>
>>
>> Regards,
>>
>> James
>>
>>
>


-- 
*James Roper*
*Senior Octonaut*

Lightbend <https://www.lightbend.com/> – Build reactive apps!
Twitter: @jroper <https://twitter.com/jroper>

Re: websockets

[Chuck Davis]

James, this does not make any sense to me.  Last I heard browsers no longer
run java.  Why would JSE give a hang about designing a communications
mechanism for browsers on the client side?  That is the job of the browser
developers.  On  the other hand, JSE WebSocket is the best thing to come
along for java clients since RMI which was released I think in 1.1.  Java
WebSocket is NOT for browsersit's for Java clients    And I'm
quite pleased with the progress I've made refactoring to use the jdk9
WebSocket api together with wildfly11 on the server (and still blazingly
FAST -- seems to be snappier than jdk8 implementation).  Now if the IDEs
would get in gear to fully implement jdk9 I'd be in fat city.

And my thanks to everyone who has participated in this particular
conversation on this list and all the expertise that has been shared.

>Yes, there's no formal definition of high/low level, but in the case of
WebSockets, we can draw some lines based roughly on what web browsers
(which is the target application developer experience that >WebSockets were
designed for) support as being high level, and what the WebSocket protocol
allows beyond this as low level.

>
> Regards,
>
> James
>
>

Re: websockets

[James Roper]

Hi Simone,

Sorry, I think there's been some missed information here. I am in no way
suggesting that the API as provided currently should be changed or
replaced. What I'm suggesting is that a high level adapter be provided out
of the box that provides Reactive Streams support on top of this API. Yes,
there's no formal definition of high/low level, but in the case of
WebSockets, we can draw some lines based roughly on what web browsers
(which is the target application developer experience that WebSockets were
designed for) support as being high level, and what the WebSocket protocol
allows beyond this as low level. I think a Reactive Streams adapter on top
of the existing low level API that has feature parity with the JavaScript
API for WebSockets offered by browsers is a worthwhile thing to aim for.
Would you agree with that?

So, for example, the browser APIs offer one error callback for handling
errors from both directions. All WebSocket messages handled by the browser
APIs are discrete, fixed length in memory, any chunking is handled
transparently underneath. Developers in the browser don't have to
explicitly send or implement any sending or receiving of the close
messages, they just handle the high level concept of the socket being
closed, and the browser implements the handshake underneath transparently.

I'd say WebSocket proxies would be completely out of scope for a high level
API, they definitely would want to deal with splitting messages, custom
close handling, very precise error semantics etc, and such a use case would
be well served by the current low level API.

Regards,

James

On 26 February 2018 at 22:15, Simone Bordet <simone.bor...@gmail.com> wrote:

> James,
>
> On Mon, Feb 26, 2018 at 4:37 AM, James Roper <ja...@lightbend.com> wrote:
> > On the topic of error handling. A high level API doesn't need to report
> each
> > individual error with sending.
>
> Depends on what you mean by "high level APIs"... I guess there is no
> formal definition of that, as one can even think that the Java socket
> APIs are high level.
>
> > So firstly, it is impossible to report *all*
> > errors with sending, since it's impossible to know, once you send a
> message
> > across the network, whether it got there or not. So if an application
> > developer has a requirement to handle errors in sending (and
> realistically,
> > there is never a business require to handle just the errors that can be
> > detected in sending, it's always about handling all errors), do we expect
> > them to implement that logic in two places, both on the sending side to
> > handle errors that can be detected on the sending side, and then
> > additionally write logic to handle errors that can't be detected on the
> > sending side (such as, for example, having the remote side send ACKs, and
> > track the ACKs on the receiving end)? I doubt it, having to handle logic
> on
> > both sides is going to mean the application developer has to spend more
> time
> > implementing boiler plate that could otherwise be spent solving their
> actual
> > business problem.
>
> I think you are oversimplifying this.
>
> If my business is to write a WebSocket Proxy, I may totally want to
> care about writing code that handles errors differently whether they
> happened on the read side or on the write side, if just for logging
> reasons.
>

> If my business is to write a protocol on top of WebSocket, I may want
> to use custom WebSocket error codes (depending on errors) when I close
> the connection.
> I may want to send a WS message the moment I receive a WebSocket
> "close" event from a peer.
>
> Sure, maybe in 80% of the cases read and write errors are handled the
> same way, but as an application writer I would like to have the choice
> between using an API that is "lower" level and allows me full
> flexibility with respect to the WebSocket protocol, and an opinionated
> (correctly opinionated in the majority of the cases) framework that
> for some case reduces the API surface at the cost of something else
> like flexibility, increased allocation, ecc.
> Both have their use cases.
>
> I believe the JDK net group always stated they went for the first choice.
> The lack of the second choice is under discussion, and as I have
> already stated may be a matter of resources.
>
> > If an application developer needs to handle errors in
> > sending messages, it would be much simpler for them to treat both types
> of
> > errors in the same way. And hence, a high level API should expose
> detectable
> > errors in sending on the receiving end, by cancelling upstream,
> terminating
> > the WebSocket with an error, and then emitting the error in the
> downstream

Re: websockets

[Simone Bordet]

James,

On Mon, Feb 26, 2018 at 4:37 AM, James Roper  wrote:
> On the topic of error handling. A high level API doesn't need to report each
> individual error with sending.

Depends on what you mean by "high level APIs"... I guess there is no
formal definition of that, as one can even think that the Java socket
APIs are high level.

> So firstly, it is impossible to report *all*
> errors with sending, since it's impossible to know, once you send a message
> across the network, whether it got there or not. So if an application
> developer has a requirement to handle errors in sending (and realistically,
> there is never a business require to handle just the errors that can be
> detected in sending, it's always about handling all errors), do we expect
> them to implement that logic in two places, both on the sending side to
> handle errors that can be detected on the sending side, and then
> additionally write logic to handle errors that can't be detected on the
> sending side (such as, for example, having the remote side send ACKs, and
> track the ACKs on the receiving end)? I doubt it, having to handle logic on
> both sides is going to mean the application developer has to spend more time
> implementing boiler plate that could otherwise be spent solving their actual
> business problem.

I think you are oversimplifying this.

If my business is to write a WebSocket Proxy, I may totally want to
care about writing code that handles errors differently whether they
happened on the read side or on the write side, if just for logging
reasons.

If my business is to write a protocol on top of WebSocket, I may want
to use custom WebSocket error codes (depending on errors) when I close
the connection.
I may want to send a WS message the moment I receive a WebSocket
"close" event from a peer.

Sure, maybe in 80% of the cases read and write errors are handled the
same way, but as an application writer I would like to have the choice
between using an API that is "lower" level and allows me full
flexibility with respect to the WebSocket protocol, and an opinionated
(correctly opinionated in the majority of the cases) framework that
for some case reduces the API surface at the cost of something else
like flexibility, increased allocation, ecc.
Both have their use cases.

I believe the JDK net group always stated they went for the first choice.
The lack of the second choice is under discussion, and as I have
already stated may be a matter of resources.

> If an application developer needs to handle errors in
> sending messages, it would be much simpler for them to treat both types of
> errors in the same way. And hence, a high level API should expose detectable
> errors in sending on the receiving end, by cancelling upstream, terminating
> the WebSocket with an error, and then emitting the error in the downstream
> onError.
>
> When it comes to protocol specific concerns - a high level API should not
> expose that to developers.  A developer should not be allowed to send a text
> message between two split binary frames. A developer should not be
> responsible for implementing the closing handshake protocol. That is the
> responsibility of a high level API. WebSocket close messages are designed
> such that high level APIs don't need to expose them to end developers - you
> have one status for success, and then many statuses for close. This means,
> an application developer can signal a normal close by simply terminating the
> stream, and can signal any of the other closes by terminating with an error
> that gets mapped to an error status code - likewise, closes received can be
> mapped in the same way.  And, most of the error codes are for protocol level
> errors anyway that a user should never be generating and can't really
> handle. So I don't think, if providing a high level API, it makes any sense
> to expose close messages or the close handshake to application developers -
> the WebSocket protocol by design is not meant to be used that way.

I don't understand what you're saying here.
You want to notify application code if the other peer closed the connection.
If you do, then you're exposing the application code to the WebSocket
close event, which is part of the WebSocket close handshake.
An application can use custom WebSocket error codes, or may want to
reply with a different error code, or may want to send a last message
with information of what has been already processed so far (similar to
HTTP/2 last stream id).
All that can only be provided by application code so the API must
allow for that.
Again, if you want to write an opinionated framework that covers 80%
of the cases in a simpler way, great, but I'd like to have other
choices too.

> For fragmented frames, once again, I don't think the designers of the
> WebSocket protocol ever intended that this protocol level detail would ever
> be exposed to application developers. My reading of the spec is that the
> reason fragmentation is allowed 

Re: websockets

[James Roper]

On the topic of error handling. A high level API doesn't need to report
each individual error with sending. So firstly, it is impossible to report
*all* errors with sending, since it's impossible to know, once you send a
message across the network, whether it got there or not. So if an
application developer has a requirement to handle errors in sending (and
realistically, there is never a business require to handle just the errors
that can be detected in sending, it's always about handling all errors), do
we expect them to implement that logic in two places, both on the sending
side to handle errors that can be detected on the sending side, and then
additionally write logic to handle errors that can't be detected on the
sending side (such as, for example, having the remote side send ACKs, and
track the ACKs on the receiving end)? I doubt it, having to handle logic on
both sides is going to mean the application developer has to spend more
time implementing boiler plate that could otherwise be spent solving their
actual business problem.  If an application developer needs to handle
errors in sending messages, it would be much simpler for them to treat both
types of errors in the same way. And hence, a high level API should expose
detectable errors in sending on the receiving end, by cancelling upstream,
terminating the WebSocket with an error, and then emitting the error in the
downstream onError.

When it comes to protocol specific concerns - a high level API should not
expose that to developers.  A developer should not be allowed to send a
text message between two split binary frames. A developer should not be
responsible for implementing the closing handshake protocol. That is the
responsibility of a high level API. WebSocket close messages are designed
such that high level APIs don't need to expose them to end developers - you
have one status for success, and then many statuses for close. This means,
an application developer can signal a normal close by simply terminating
the stream, and can signal any of the other closes by terminating with an
error that gets mapped to an error status code - likewise, closes received
can be mapped in the same way.  And, most of the error codes are for
protocol level errors anyway that a user should never be generating and
can't really handle. So I don't think, if providing a high level API, it
makes any sense to expose close messages or the close handshake to
application developers - the WebSocket protocol by design is not meant to
be used that way.

For fragmented frames, once again, I don't think the designers of the
WebSocket protocol ever intended that this protocol level detail would ever
be exposed to application developers. My reading of the spec is that the
reason fragmentation is allowed is to allow endpoints to work with fixed
buffer sizes, allowing messages to be split if they choose. In fact, this
is exactly what the RFC says:

The primary purpose of fragmentation is to allow sending a message
> that is of unknown size when the message is started without having to
> buffer that message.  If messages couldn't be fragmented, then an
> endpoint would have to buffer the entire message so its length could
> be counted before the first byte is sent.  With fragmentation, a
> server or intermediary may choose a reasonable size buffer and, when
> the buffer is full, write a fragment to the network.


https://tools.ietf.org/search/rfc6455#section-5.4

This is a protocol level detail, not something that is supposed to be
exposed to application developers. It is reasonable to expect that an
implementation will buffer and put the fragments back together, up to a
configured buffer size, for application consumption. If a message does
exceed that buffer size, then the implementation can close the socket with
the close code designed exactly for that purpose - 1009.

I don't think a high level API should seek to make every feature of
WebSockets available just because it's possible in the protocol - I think
that would be over-engineering, over complicating things for application
developers. I think a high level API should be focused on how the protocol
is intended to be used by application developers.

On 24 February 2018 at 22:55, Chris Hegarty <chris.hega...@oracle.com>
wrote:

> Rossen,
>
> On 23 Feb 2018, at 21:15, Rossen Stoyanchev <rstoyanc...@pivotal.io>
> wrote:
>
> hi Pavel,
>
> On Thu, Feb 22, 2018 at 8:43 AM, Pavel Rappo <pavel.ra...@oracle.com>
> wrote:
>
>>
>> 1. If there is no error reporting, then how would the user know if there's
>> something wrong with the data they send? I'm talking about
>> incomplete/malformed
>> UTF-8, fragments intermixing, sending messages after a Close message has
>> been
>> sent, etc. How should the user troubleshoot a situation where the
>> WebSocket's
>> internal Subscriber cancels the subscription unexpect

Re: websockets

[Chris Hegarty]

Rossen,

> On 23 Feb 2018, at 21:15, Rossen Stoyanchev  wrote:
> 
> hi Pavel,
> 
> On Thu, Feb 22, 2018 at 8:43 AM, Pavel Rappo  > wrote:
> 
> 1. If there is no error reporting, then how would the user know if there's
> something wrong with the data they send? I'm talking about 
> incomplete/malformed
> UTF-8, fragments intermixing, sending messages after a Close message has been
> sent, etc. How should the user troubleshoot a situation where the WebSocket's
> internal Subscriber cancels the subscription unexpectedly? We can't perform 
> all
> the checks during a message creation. Some messages could not be checked up
> until they appear in some context. For example, a text fragment is a okay per
> se. But if it appears in the midst of a fragmented binary message, it is not
> okay. Should the API not allow fragmented (partial) messages then? Or should 
> we
> provide error reporting out-of-band in relation to Flow? If we allow only 
> whole
> messages, then we are not flexible enough to support streaming. In other words
> the case of an arbitrarily large message that does not have to be assembled
> prior to consumption.
> 
> Charset encoding issues aside, are these represented with specific 
> exceptions, and how much can the application do with them?

An application can handle, and possibly recover.

From WebSocketMessage [1]:

/**
 * Return the message payload as UTF-8 text. This is a useful for text
 * WebSocket messages.
 */
public String getPayloadAsText() {
byte[] bytes = new byte[this.payload.readableByteCount()];
this.payload.read(bytes);
return new String(bytes, StandardCharsets.UTF_8);
}

This seems woefully inadequate, and underspecified. Not to mention what
happens if the message is not text. The Java SE API attempts to avoid such.


> I don't know the fragments intermixing issue but can the application recover 
> from that?

With the Java SE API, yes.

> It is of course more useful for the API to expose error information for each 
> sent message, but if those are low level errors, treated as terminal, then 
> only the first one practically matters.

That is not always the case. And this is a low-level API.

> I'm unsure about the exact meaning of fragments​. JSR-356 has partial 
> messages, which if I recall were explicitly called out in the spec as not 
> being different from (and not 1-for-1 with) WebSocket fragments. I'll assume 
> that you mean the same (i.e. the send methods that take an extra isLast 
> flag). WebSocket is not an application-level protocol and in my experience, 
> streaming by splitting large content, is a higher level concern. For example 
> the JavaScript client for STOMP over WebSocket some time ago started 
> splitting larger STOMP messages along 16K boundaries to work better with the 
> default buffer sizes of most servers. I'm not saying such feature shouldn't 
> be present but I don't think it's a must.

Why preclude it, when it is not necessary to do so?

> If required, one way to represent it is to allow writing with a 
> Publisher.

Yuck!

> 2. How should the user close their side of the communication? There seems to 
> be
> a mismatch in the protocols. WebSocket is bidirectional and relies on Close
> messages. Flow is unidirectional-ish and relies on cancellation through
> onComplete/onError and cancel signals. If the user has no more messages to
> publish, the user may wish to send a Close message with a status code and a
> reason. If instead we rely on onComplete/onError being called then we loose 
> the
> ability to provide these extra pieces of information. What should we choose
> here? To not allow a Close message to be sent or to require the user to signal
> onComplete() after a Close message has been signalled through onNext(). Maybe
> there should not be a representation of a Close message at all. Maybe we 
> should
> leave it as a protocol low-level detail. But then, how should the API deliver
> the received status code and the reason of a Close message it receives? Should
> it translate the received Close message to onComplete()/onError() depending on
> this message's contents?
> 
> I think the API can provide such a close(CloseStatus) method. That would 
> result in a cancellation of the write Publisher.
>  
> 3. WebSocket protocol mandates a peer should send a Close message as soon as
> practical after receiving one. In the current API it is implemented using a
> CompletionStage returned from the WebSocket.Listener.onClose method. How 
> should
> we implement this using Flow?
> 
> It is hard to discuss on this level of detail without a POC. 

We have gone through many POC’s over the past 3 or more years, bringing
us to where we are today. Much of this work and is archived on the net-dev
mailing list, if you wish to review it. It’s the only way to validate API
suggestions.

> I'd assume the API exposes a "read" 

Re: websockets

[Rossen Stoyanchev]

hi Pavel,

On Thu, Feb 22, 2018 at 8:43 AM, Pavel Rappo  wrote:

>
> 1. If there is no error reporting, then how would the user know if there's
> something wrong with the data they send? I'm talking about
> incomplete/malformed
> UTF-8, fragments intermixing, sending messages after a Close message has
> been
> sent, etc. How should the user troubleshoot a situation where the
> WebSocket's
> internal Subscriber cancels the subscription unexpectedly? We can't
> perform all
> the checks during a message creation. Some messages could not be checked up
> until they appear in some context. For example, a text fragment is a okay
> per
> se. But if it appears in the midst of a fragmented binary message, it is
> not
> okay. Should the API not allow fragmented (partial) messages then? Or
> should we
> provide error reporting out-of-band in relation to Flow? If we allow only
> whole
> messages, then we are not flexible enough to support streaming. In other
> words
> the case of an arbitrarily large message that does not have to be assembled
> prior to consumption.
>

Charset encoding issues aside, are these represented with specific
exceptions, and how much can the application do with them? I don't know the
fragments intermixing issue but can the application recover from that? It
is of course more useful for the API to expose error information for each
sent message, but if those are low level errors, treated as terminal, then
only the first one practically matters.

I'm unsure about the exact meaning of fragments​. JSR-356 has partial
messages, which if I recall were explicitly called out in the spec as not
being different from (and not 1-for-1 with) WebSocket fragments. I'll
assume that you mean the same (i.e. the send methods that take an extra
isLast flag). WebSocket is not an application-level protocol and in my
experience, streaming by splitting large content, is a higher level
concern. For example the JavaScript client for STOMP over WebSocket some
time ago started splitting larger STOMP messages along 16K boundaries to
work better with the default buffer sizes of most servers. I'm not saying
such feature shouldn't be present but I don't think it's a must. If
required, one way to represent it is to allow writing with a
Publisher.


> 2. How should the user close their side of the communication? There seems
> to be
> a mismatch in the protocols. WebSocket is bidirectional and relies on Close
> messages. Flow is unidirectional-ish and relies on cancellation through
> onComplete/onError and cancel signals. If the user has no more messages to
> publish, the user may wish to send a Close message with a status code and a
> reason. If instead we rely on onComplete/onError being called then we
> loose the
> ability to provide these extra pieces of information. What should we choose
> here? To not allow a Close message to be sent or to require the user to
> signal
> onComplete() after a Close message has been signalled through onNext().
> Maybe
> there should not be a representation of a Close message at all. Maybe we
> should
> leave it as a protocol low-level detail. But then, how should the API
> deliver
> the received status code and the reason of a Close message it receives?
> Should
> it translate the received Close message to onComplete()/onError()
> depending on
> this message's contents?
>

I think the API can provide such a close(CloseStatus) method. That would
result in a cancellation of the write Publisher.


> 3. WebSocket protocol mandates a peer should send a Close message as soon
> as
> practical after receiving one. In the current API it is implemented using a
> CompletionStage returned from the WebSocket.Listener.onClose method. How
> should
> we implement this using Flow?
>

It is hard to discuss on this level of detail without a POC.

I'd assume the API exposes a "read" Publisher, a way to send
with a "write" Publisher and some Handler/Listener that returns
Publisher (or CompletionStage) for when handling is complete. The
"read" Publisher would complete when the server closes. The Publisher
from the Handler/Listener would complete when all is handled.



> Sure one may recycle buffers transparently. But only if the
> buffer type supports it. io.netty.buffer.PooledByteBufAllocator provides
> instances of io.netty.buffer.ByteBuf. Not java.nio.ByteBuffer. The latter
> one
> does not have any release/free/dispose/etc. methods and semantics.
>

​If the API exposed some Consumer callback to be applied to
buffers that have been written, then such a pool could also be implemented
externally.
​​

> P.S. How may purely RS WebSocket APIs are there? If there are some, we can
> probably
> learn from their experience and make things a bit better. Otherwise I
> don't see
> why we should create one more of them. Would you rather have something
> low-level, you could then implement your high-level API that is most
> relevant to
> your case?
>

You can see the reactive WebSocket API 

Re: websockets

[Pavel Rappo]

Hello Rossen,

> On 22 Feb 2018, at 02:51, Rossen Stoyanchev  wrote:
> 
> ​​hi,
> 
> > Simone Bordet:
> > As the WebSocket APIs were moved to incubation and are now slated
> > for JDK 11, it may be worth considering again whether it's the
> > case to provide a Flow-based APIs.
> 
> I second that. There were two main issues pointed out in this thread:
> 
> > The first one is how to communicate errors back to the user's
> > code that publishes messages to WebSocket.
> 
> Isn't it mostly low level, transport errors about which the application can't 
> do much about, but clean up and close the connection? Arguably a cancellation 
> is adequate, or do you have more specific errors and cases in mind? 
> Alternatively, the JSR-356 WebSocket API provides a single onError callback 
> on the Endpoint type. I haven't come across any limitations around that.
> 

I can't see how Flow (RS) can be seamlessly married with this low-level
WebSocket API. Maybe you might help.

1. If there is no error reporting, then how would the user know if there's
something wrong with the data they send? I'm talking about incomplete/malformed
UTF-8, fragments intermixing, sending messages after a Close message has been
sent, etc. How should the user troubleshoot a situation where the WebSocket's
internal Subscriber cancels the subscription unexpectedly? We can't perform all
the checks during a message creation. Some messages could not be checked up
until they appear in some context. For example, a text fragment is a okay per
se. But if it appears in the midst of a fragmented binary message, it is not
okay. Should the API not allow fragmented (partial) messages then? Or should we
provide error reporting out-of-band in relation to Flow? If we allow only whole
messages, then we are not flexible enough to support streaming. In other words
the case of an arbitrarily large message that does not have to be assembled
prior to consumption.

2. How should the user close their side of the communication? There seems to be
a mismatch in the protocols. WebSocket is bidirectional and relies on Close
messages. Flow is unidirectional-ish and relies on cancellation through
onComplete/onError and cancel signals. If the user has no more messages to
publish, the user may wish to send a Close message with a status code and a
reason. If instead we rely on onComplete/onError being called then we loose the
ability to provide these extra pieces of information. What should we choose
here? To not allow a Close message to be sent or to require the user to signal
onComplete() after a Close message has been signalled through onNext(). Maybe
there should not be a representation of a Close message at all. Maybe we should
leave it as a protocol low-level detail. But then, how should the API deliver
the received status code and the reason of a Close message it receives? Should
it translate the received Close message to onComplete()/onError() depending on
this message's contents?

3. WebSocket protocol mandates a peer should send a Close message as soon as
practical after receiving one. In the current API it is implemented using a
CompletionStage returned from the WebSocket.Listener.onClose method. How should
we implement this using Flow?

Having said that, it's either (but not both):

  a) We ditch low-level API (to Ping/Pong/Close and (possibly) incomplete
 messages and start using Flow, or
  b) We leave the API as it is and instead provide a Flow-adaptor suitable
 for the anticipated most common use case. As for now let's not dwell into
 details of who and when provides this adaptor.

> 
> 
> 
> James Roper​ mentioned this already, but the Netty PooledByteBufAllocator 
> demonstrates it's feasible to recycle buffers transparently without the need 
> for a per-message confirmation. I don't think the argument holds that not 
> having per message completion signals forces unnecessary allocations, garbage 
> creation and copying.
> 

Sure one may recycle buffers transparently. But only if the
buffer type supports it. io.netty.buffer.PooledByteBufAllocator provides
instances of io.netty.buffer.ByteBuf. Not java.nio.ByteBuffer. The latter one
does not have any release/free/dispose/etc. methods and semantics.

> 
> There were a "dozen of smaller issues" mentioned. However there were no links 
> to earlier discussions, so I'm not sure what those were. 
> 

If you want you can spend some time tracking the evolution of the API:

http://mail.openjdk.java.net/pipermail/net-dev/2015-September/009079.html
http://mail.openjdk.java.net/pipermail/net-dev/2015-October/009223.html
http://mail.openjdk.java.net/pipermail/net-dev/2015-October/009259.html
http://mail.openjdk.java.net/pipermail/net-dev/2016-March/009616.html
http://mail.openjdk.java.net/pipermail/net-dev/2016-April/009632.html
http://mail.openjdk.java.net/pipermail/net-dev/2016-April/009702.html
http://mail.openjdk.java.net/pipermail/net-dev/2016-May/009749.html
  

Re: websockets

[Rossen Stoyanchev]

​​hi,

> Simone Bordet:
> As the WebSocket APIs were moved to incubation and are now slated
> for JDK 11, it may be worth considering again whether it's the
> case to provide a Flow-based APIs.

I second that. There were two main issues pointed out in this thread:

> The first one is how to communicate errors back to the user's
> code that publishes messages to WebSocket.

Isn't it mostly low level, transport errors about which the application
can't do much about, but clean up and close the connection? Arguably a
cancellation is adequate, or do you have more specific errors and cases in
mind? Alternatively, the JSR-356 WebSocket API provides a single onError
callback on the Endpoint type. I haven't come across any limitations around
that.

> The second issue is how to communicate completion signals from
> processing individual messages.

James Roper​ mentioned this already, but the Netty PooledByteBufAllocator
demonstrates it's feasible to recycle buffers transparently without the
need for a per-message confirmation. I don't think the argument holds that
not having per message completion signals forces unnecessary allocations,
garbage creation and copying.

There were a "dozen of smaller issues" mentioned. However there were no
links to earlier discussions, so I'm not sure what those were.

The main purpose of Reactive Streams is to connect components from
different libraries. The currently proposed WebSocket API cannot be easily
integrated into a Reactive Streams pipeline. One would have to build a
bridge to Reactive Streams first and I don't think that should be left as a
separate exercise, since Flow is in the JDK and it's the reason why it was
introduced, to provide a common contract for such scenarios.

> Another point that makes me feeling uncertain is the good comment by
> Konrad Malawski about the design of Akka Streams versus e.g. Spring's
> Flowable, where the former does not implement RS or Flow interfaces,
> while the latter does.

One minor clarification first. By "Spring's Flowable" I presume you mean
the Flux and Mono types from the Reactor project? Reactor does not depend
on any Spring projects, and it is a lower level project. It's directly
comparable to RxJava, in fact it implements the same ReactiveX patterns,
but more explicitly targeting server-side Java applications vs Android.

To the point, I don't think the Flow vs RS contracts should a cause for
hesitation. HttpClient already uses the Flow contract and that's the
obvious choice for WebSocket too. More broadly, the JDK Flow and the RS
contracts are identical, and have identical semantics, so adapting from one
to the other is trivial. In most cases applications are not consuming these
contracts directly, which is also Konrad's point, so practically speaking I
don't see much dichotomy there.​

Regards,
Rossen

​

Re: websockets

[James Roper]

If it's a question of resources for building a separate Flow based API on
top of the existing API provided, then perhaps we could help? I could
certainly implement it outside the JDK in the coming months, would the HTTP
client team be willing to consider folding it in as a utility alongside the
WebSocket API in the JDK? I know that's not a zero resource thing, but
assuming I can produce something reasonable, reviewing it should be cheaper
than implementing it. Also I don't know if there's any work being done on a
TCK for this, and whether inclusion of this in the JDK would require a TCK
to verify it.

On 20 February 2018 at 08:41, Simone Bordet  wrote:

> James (hi! :)
>
> On Fri, Feb 16, 2018 at 12:55 AM, James Roper  wrote:
> > The question for me then is whether there is value in introducing an
> > additional higher level API (ie, a reactive streams based API) for the
> > purposes of integration with other libraries that provide RS support.
>
> I have mixed feeling about this.
>
> I first tried to push the WebSocket APIs to be Flow-based when they
> were slated for JDK 9 official inclusion.
> As the WebSocket APIs were moved to incubation and are now slated for
> JDK 11, it may be worth considering again whether it's the case to
> provide a Flow-based APIs.
>
> However, Pavel's point about the limits of the Flow APIs still stand,
> so a Flow API may be useful, but less generic.
>
> Another point that makes me feeling uncertain is the good comment by
> Konrad Malawski about the design of Akka Streams versus e.g. Spring's
> Flowable, where the former does not implement RS or Flow interfaces,
> while the latter does.
> Now that we have the dichotomy of RS and JDK's Flow, it is easy for
> Akka Stream implement methods such as asRSPublisher() and
> asFlowPublisher() so that they work in both environments (with - yuck
> - multi-release jars).
> It would be harder for Spring's Flowable to provide support for both
> RS and Flow.
>
> So at the end I would like to see a *separate* Flow API on top of the
> current WebSocket API, but who's going to implement it ?
> Would be great to standardize a community effort, but even better if
> the JDK provided it.
>
> > Introducing another API, regardless of what that API is, obviously has
> > drawbacks as it increases the surface area of the API, and thus
> complexity,
> > and also potentially introduces confusion for application developers as
> they
> > now have two different ways of achieving things, and they need to know
> which
> > to use when. So, the value of that API must be high enough to outweigh
> those
> > drawbacks. I strongly believe that a Reactive Streams based API does add
> > enough value to do that - and the post that I shared in my previous email
> > demonstrates how an ecosystem where reactive streams is embraced can
> provide
> > a lot of value to application developers. So I'm happy to talk more here
> > about use cases and code samples etc if you don't agree here.
>
> I personally don't question the value of a RS or Flow API.
> It's just a matter of resources - we can ask Oracle to do it, but if
> they don't have resources to do it, then the road ahead is
> complicated.
> I see better a Flow wrapper of the JDK WebSocket APIs under the
> Reactive Extension (https://github.com/Reactive-Extensions) or
> Reactive Streams (https://github.com/Reactive-Streams) umbrella.
> That would perhaps have less resource issues and still give a unique
> library rather than a gazillion similar wrappers.
>
> > For signalling errors upstream, in all the cases that I've worked with,
> > we've never found a reason that sending the error downstream, and just
> > cancelling upstream, is a problem. At the end of the day, there is only
> one
> > thing that can fail, the TCP connection, and if it fails, both upstream
> and
> > downstream will fail - there is no such thing as a half failure in TCP.
>
> HttpServlet.service() allows to throw exceptions that the Servlet
> container must handle by running the error page algorithm, which is
> user code.
> A Subscriber (provided by a library) encounters an exception while
> performing its job.
> Since it comes from a library, the library itself does not know how to
> handle the error because it does not know in what environment it is
> being used.
> The common solution I saw in RS libraries for this is to provide
> library specific APIs that registers a callback that is invoked in
> case of errors.
> Another common solution is to have an inverse RS that relays errors in
> the opposite direction.
> With this library specific APIs, an app running in a Servlet Container
> can trigger the error page algorithm depending on the error provided
> by the library.
> In this way, the error is reported upstream but using an out-of-band
> channel specific to each RS library.
> And it is *necessary* to notify the error upstream.
>
> Now we can argue forever about whether the mechanism should be part of

Re: websockets

[James Roper]

Hi Pavel,

Sorry for taking a while to respond, I've been travelling and wanted to
take the time to properly understand the concerns you've raised.

The reasoning you've stated for the current API design is all sound, and
obviously the choice to go with that design and not others depends on the
goals of the API. I don't think I'm in a position to be able to disagree
with the goals that you've set, I'm sure you've done a lot of research,
talked to a lot of people and got a lot of feedback. So I'm happy to accept
that the current API as it stands is a good one and should stay as is.

The question for me then is whether there is value in introducing an
additional higher level API (ie, a reactive streams based API) for the
purposes of integration with other libraries that provide RS support.
Introducing another API, regardless of what that API is, obviously has
drawbacks as it increases the surface area of the API, and thus complexity,
and also potentially introduces confusion for application developers as
they now have two different ways of achieving things, and they need to know
which to use when. So, the value of that API must be high enough to
outweigh those drawbacks. I strongly believe that a Reactive Streams based
API does add enough value to do that - and the post that I shared in my
previous email demonstrates how an ecosystem where reactive streams is
embraced can provide a lot of value to application developers. So I'm happy
to talk more here about use cases and code samples etc if you don't agree
here.

To address the issues you raised - I think when providing reactive streams
integration, we need to accept that it is a high level API, and there can
be costs associated with it - performance costs, in some cases
functionality costs, etc. This is the cost of abstracting a problem away so
that many different libraries can reliably integrate with each other -
there are many libraries that work with reusable buffers, but there is no
standard for how the lifecycle of those buffers are managed, in some cases
they get reused immediately, in other cases, for example, in Netty, they
use reference counting, and transparently return the buffer to a pool when
the reference count reaches zero. The Reactive Streams spec specifically
decided not to deal with this side of integrating streaming libraries,
which means either that some out of band mechanism needs to be provided for
signalling the end of use of buffers (which puts constraints on interop
because not every RS sink/source that you might plumb in would support
integrating with that out of band mechanism), or that reusable buffers
cannot be used - in general, the latter would probably be preferred, as
high level concurrency integration code generally requires messages sent
between different components to be immutable anyway.

So, I would say that a reactive streams interface would copy the reusable
buffer into a read only immutable buffer, and that this choice of decreased
performance would be a trade off given to the application developer when
they choose between using the RS API or writing their own code to use the
proprietary API.

For signalling errors upstream, in all the cases that I've worked with,
we've never found a reason that sending the error downstream, and just
cancelling upstream, is a problem. At the end of the day, there is only one
thing that can fail, the TCP connection, and if it fails, both upstream and
downstream will fail - there is no such thing as a half failure in TCP. So
the error sent down, and the error sent up, will always represent the same
failure of the TCP socket, and probably will always be exactly the same
exception. So, if you send errors both up and down, then the application
has a problem - which one should it handle? Should it handle both? The end
effect of handling both, at a minimum, will usually mean every failure of a
WebSocket connection will result in the error appearing twice in an
applications logs, once for downstream, once for upstream, which is not
helpful because there was only actually one error. At worst, it means for
example if you have code to restart a WebSocket connection, without
carefully written concurrent code, you could end up double restarting the
streams. By only sending errors through the Subscriber interface, ie
downstream, you simplify the application handling of errors, because the
application only needs to handle the error in one place, rather than having
to make a choice between handling it in one place or another, and getting
that choice right. The issue you raised with this meaning that this means
there will be coupling between the subscriber and publisher sides - while
that's correct, it's not the error handling that would cause that. The
publisher and subscriber side are always intrinsically linked, because they
are bound to the same TCP connection - they live and die based on that TCP
connection. If one side fails, so does the other side. If one side starts,
so does the otherside. 

Re: websockets

[Chuck Davis]

Thanks, Pavel, for the additional information.

Text payloads will be easily handled.  ByteBuffers not so much
apparently.  Major issues I see now:  they can't grow and all the
methods I'd want to use are abstract, optional or both.  So each
implementation I run on will have to be tested to see if the JVM even
implements the methods...such basic things as get() and put().  So I'm
researching to find a good buffering mechanism.  If you have
suggestion (or anyone else lurking here) I'm wide open to suggestions
how best to do buffering for ByteBuffers so that serialized objects
can be deserialized.  Didn't have this issue with jdk8 WebSocket since
I only got complete messages to decode.  Backpressure was not an issue
in my development environment but I have no problem understanding why
it might be needed.

With the test database I'm currently working with a typical payload
will be about 10k in size (many smaller but some larger).  I'm going
to have to devise a GOOD way to buffer these partial messages into a
whole buffer/stream before I submit to an ObjectInputStream to
deserialize the objects. Of course when I connect to a real, live
database a payload of 100k or more will not be an unusual event.

I am well aware I'm not the sharpest knife in the drawer so welcome
strategies of which objects to use, and how, to accomplish the task of
putting Humpty Dumpty back together again after having been broken
into many pieces of ByteBuffer.

CD

On Wed, Feb 14, 2018 at 10:38 AM, Pavel Rappo  wrote:
>

> It does however seem that optimization-wise there is not much to be gained 
> from
> doing this. As different parts of a whole message (i.e. payloads of WebSocket
> frames) do not occupy contiguous parts of the underlying memory. In order to 
> be
> concatenated, payload pieces would need to be cut out of their frames first.
> Which means copying. In this context it might have been better if we decided 
> to
> use ByteBuffer[] instead of ByteBuffer in the first place. But that seemed 
> like
> an overkill back then.
>

Re: websockets

[Pavel Rappo]

> On 12 Feb 2018, at 19:29, Chuck Davis  wrote:
> 
> 
> 
> There is never a way to tell how much data may be
> returned.
> 

Reactive Streams speak in terms of numbers of invocations to the onNext method.
Not the amount of data. (Please disregard if that's not what you meant.)

> 
> 
> I am under the impression that this new API requires the developer to
> buffer the messages until all the data is complete. As I recall (it's
> been a few months since I implemented) decoders handled that in the
> old API and if I am correct that is going to put a lot more work on
> the plate of developers.  If my understanding is correct I'd suggest
> listeners should do the buffering and deliver a complete message.  If
> it's written once in the implementation it would save thousands of
> developer hours recreating the same wheel.  At least this approach
> makes sense in my use case.  Partial pieces of serialized objects are
> useless.
> 

Different developers have different requirements. Some would benefit from having
the ability to receive partial messages. If this API only provided whole message
receiving, it would not be possible to satisfy those developers.
If in your case it makes no sense to use partial messages, you can still use the
API. It will only require some extra work though. In the simplest case the
amount of work is not that big really:

   @Override
   public CompletionStage onText(WebSocket webSocket,
CharSequence message,
WebSocket.MessagePart part) {
   webSocket.request(1);
   String str = null;
   switch (part) {
   case FIRST:
   case PART:
   text.append(message);
   return null;
   case LAST:
   text.append(message);
   str = text.toString();
   text.setLength(0);
   break;
   case WHOLE:
   str = message.toString();
   break;
   }
   processWholeText(str);
   return null;
   }

   private void processWholeText(String string) {
   // -- your code here --
   }

However, with this approach you will need to go the extra mile if you ever need
to talk to WebSocket.request outside of the context of the associated listener.
Because the WebSocket client will continue to speak in terms of the number of
unspecified (could be whole, could be partial) messages.

We might provide this "buffering" ability internally in WebSocket in the future.
One option would be to add an intermediate method to the WebSocket.Builder such
that this method would give an extra guarantee for the attached listener:
 
   Builder receiveWholeMessages(boolean whole)

Another option would be to provide a slightly different listener that would
reflect this behaviour in a more type-safe manner (i.e. the listener won't have
those extra MessagePart arguments in the onText and onBinary methods):

   CompletableFuture buildAsync(URI uri, WholeListener listener)

And these are probably not the only possible options.

It does however seem that optimization-wise there is not much to be gained from
doing this. As different parts of a whole message (i.e. payloads of WebSocket
frames) do not occupy contiguous parts of the underlying memory. In order to be
concatenated, payload pieces would need to be cut out of their frames first.
Which means copying. In this context it might have been better if we decided to
use ByteBuffer[] instead of ByteBuffer in the first place. But that seemed like
an overkill back then.

> 
> Can you expound a little more on this:  "Finally, each of the
> listener's methods asks the user to signal when the user has processed
> the message it bears [2]."  All I see is that the method returns a
> CompletionStage which triggers a procession of CompletionStage objects
> (not sure yet how this ends except by returning a null?) and I don't
> see in the docs how this signals the listener except that the method
> exits and may be called again.  I'm sorry to be so dense here.
> 

Let me explain this the following way. With WebSocket there is an exchange of
messages. You create a new message for the WebSocket client and tell it to send
the message by calling one of the send methods. You'd probably like to know
though when the WebSocket client has actually processed the message. That's
where a CompletableFuture returned from the method comes in handy. When the
WebSocket client completes this future, you know the message has been processed.

Now consider that when the WebSocket client receives a message it calls one of
the receive methods on you. The WebSocket client needs you to tell it when you
finished processing the message it gave you. You do this by returning a
CompletionStage from the method. Once this stage reaches the WebSocket client,
the client then attaches dependant actions on this stage internally. But that's
details of the implementation, the point is the scenario is really 

Re: websockets

[James Roper]

On 12 February 2018 at 18:56, Chris Hegarty 
wrote:

> James,
>
> On 10/02/18 07:38, James Roper wrote:
>
>> ...
>>
>> https://developer.lightbend.com/blog/2018-02-06-reactive-str
>> eams-ee4j/index.html
>>
>
> Regarding: https://github.com/jroper/reactive-streams-servlet/blob/7a2a
> 651b706bb0612f6d11311e442f82ce307ed2/reactive-streams-servle
> t/src/main/java/org/reactivestreams/servlet/RequestPublisher.java
>
> If I'm not mistaken, this appears to be mainly an adapter
> from Publisher of InputStream to Publisher of ByteBuffer.
> Did you overlook HttpRequest.BodyPublisher
> fromInputStream​(Supplier streamSupplier)?
> Or did you run into some issue with it?


Hi Chris,

Not quite. RequestPublisher is an adapter of ServletInputStream to
Publisher, and the big difference here is that ServletInputStream has a
mechanism (introduced in Servlet 3.1) for doing non blocking IO, which is
handled by registering a ReadListener, which is not part of the InputStream
interface, but is part of the ServletInputStream interface.  Using this,
you use ServletInputStream.isReady to see if the stream is ready to return
data from ServletInputStream.read without blocking. If it is, you invoke
read, otherwise, you wait until ReadListener.onDataAvailable() is invoked,
and then you can invoke ServletInputStream.read without blocking again.

So the HttpRequest.BodyPublisher.fromInputStream, I presume, does blocking
IO on the InputStream (since InputStream only supports blocking IO), using
the blocking InputStream.read call, while RequestPublisher does non
blocking IO on ServletInputStream.

Cheers,

James


>
> -Chris.
>



-- 
*James Roper*
*Senior Octonaut*

Lightbend  – Build reactive apps!
Twitter: @jroper 

Re: websockets

[Chuck Davis]

Hi Pavel:

Thanks for your input.  You guys are a lot smarter than I am so I
won't pretend to have a situation you haven't thought about.

My typical use case is to fetch database data from the server using
ejb.  I return the data in serialized java objects to my JavaFX
clients (websockets are not just for browsers though few seem to
realize that!). There is never a way to tell how much data may be
returned.  It may be from a job costing system that returns thousands
of rows.  I like the idea of using websockets rather than a direct ejb
connection because I want, in the future, to add server pushes to my
application.  i.e. if some user adds a ledger account all registered
GL listeners would get the new account in their desktop.  Neither ejb
nor rmi provides that capability.  In fact, nothing else does except
websockets as far as I know.

I'm sure I can learn to use the new .request(long) feature.  Looking,
as you mentioned, at the Flow api I see where they used the
UnboundedSubscription with request(Long.MAX_VALUE) which was my
initial reaction to this new wrinkle.

I am under the impression that this new API requires the developer to
buffer the messages until all the data is complete. As I recall (it's
been a few months since I implemented) decoders handled that in the
old API and if I am correct that is going to put a lot more work on
the plate of developers.  If my understanding is correct I'd suggest
listeners should do the buffering and deliver a complete message.  If
it's written once in the implementation it would save thousands of
developer hours recreating the same wheel.  At least this approach
makes sense in my use case.  Partial pieces of serialized objects are
useless.

Can you expound a little more on this:  "Finally, each of the
listener's methods asks the user to signal when the user has processed
the message it bears [2]."  All I see is that the method returns a
CompletionStage which triggers a procession of CompletionStage objects
(not sure yet how this ends except by returning a null?) and I don't
see in the docs how this signals the listener except that the method
exits and may be called again.  I'm sorry to be so dense here.

Finally, is there a target release to graduate from incubator?  I'm
kinda stuck until the dust settles.  May it hit jdk10 or 11?
"Incubating Feature. Will be removed in a future release." sounds
so.ominous!

Again, Pavel, thanks for your response and interest.

Re: websockets

[Chris Hegarty]

James,

On 10/02/18 07:38, James Roper wrote:

...

https://developer.lightbend.com/blog/2018-02-06-reactive-streams-ee4j/index.html


Regarding: 
https://github.com/jroper/reactive-streams-servlet/blob/7a2a651b706bb0612f6d11311e442f82ce307ed2/reactive-streams-servlet/src/main/java/org/reactivestreams/servlet/RequestPublisher.java


If I'm not mistaken, this appears to be mainly an adapter
from Publisher of InputStream to Publisher of ByteBuffer.
Did you overlook HttpRequest.BodyPublisher
fromInputStream​(Supplier streamSupplier)?
Or did you run into some issue with it?

-Chris.

Re: websockets

[Pavel Rappo]

Hello James,

Thanks for the comprehensive reply to Chuck's email. Now regarding your own 
email.

> On 10 Feb 2018, at 07:38, James Roper  wrote:
> 
> 
> 
> But that brings me to a problem that I'd like to give as feedback to the 
> implementers - this API is not Reactive Streams, and so therefore can't take 
> advantage of Reactive Streams implementations, and more problematic, can't 
> interop with other Reactive Streams sinks/sources. If I wanted to stream a 
> WebSocket into a message broker that supports Reactive Streams, I can't. I 
> would definitely hope that Reactive Streams support could be added to this 
> API, at a minimum as a wrapper, so that application developers can easily 
> focus on their business problems, plumbing and transforming messages from one 
> place to another, rather than having to deal with implementing concurrent 
> code to pass messages.
> 

I totally understand your concern over the lack of a Reactive Streams interface
to this low-level API. All I can say is that it's not that we haven't tried. As
usual the devil is in the detail. There were at least a couple of major issues
we couldn't find a satisfactory solution to.

The first one is how to communicate errors back to the user's code that
publishes messages to WebSocket. This issue has been extensively discussed here 
[1].
The only signal the publisher can receive from a subscriber in the case of a
failure is a cancellation signal. After this signal the subscription is
considered cancelled. Now, there are different solutions to this problem, but
none of the ones we looked at seemed comprehensive. For example, such errors
can be propagated by the WebSocket to the user's subscriber receiving messages.
In which case WebSocket input and output will become tied and the WebSocket
client will seem to require both the publisher and the subscriber to be present
at the same time.

The second issue is how to communicate completion signals from processing
individual messages. That might be handy in the case the implementation or the
application decide to recycle buffers they use. With a naive RS interface to
WebSocket all the user's publisher can receive from the WebSocket's subscriber
is a request for a number of extra messages. Using only this number the
publisher cannot deduce which of the previously published messages have been
actually sent. This problem also has a number of solutions. One of which would
be to use more streams. An extra publisher WebSocket would use to emit outcomes
of sending messages and an extra subscriber WebSocket would use to receive
signals from the user once a message has been received. To be honest it looks
cumbersome.

There are also a dozen of smaller issues that have to be resolved before
creating a well-defined RS interface to WebSocket.

> It may well require wrapping messages in a high level object - text, binary, 
> ping, pong, etc, to differentiate between the message types.
> 

We went great lengths in supporting different kinds of requirements in this API.
One of the such requirements that arose from a number of discussions on this
mailing list was not to force unnecessary allocations, garbage creation and
copying. To this end we utilised completion stages for send/receive operations 
[2].
We abandoned types for messages and used a method-naming scheme instead (I
believe with RS the equivalent would be to provide a stream per message type).
Even WebSocket.Listener was designed such that one instance could (if required)
service many WebSocket instances. That's the reason each method in Listener has
a WebSocket argument.

> 
> 
> https://developer.lightbend.com/blog/2018-02-06-reactive-streams-ee4j/index.html
> 

Thanks for the link.

That said, I think once we have satisfactory answers to the questions above,
we might provide an RS adapter you were talking about to this low-level API.

Meanwhile it is possible to built one already, on top of the existing API even
though it is not the same as RS, semantically they are very similar.

Thanks,
-Pavel

---
[1] https://github.com/reactive-streams/reactive-streams-jvm/issues/271
[2] At one point we were even considering using callbacks similar to
java.nio.channels.CompletionHandler instead of 
java.util.concurrent.CompletionStage
for both sending and receiving messages. All this is for the sake of not
creating extra objects when this is considered expensive. Who knows, we
might come back to retrofit the API and add this capability later.

Re: websockets

[Viktor Klang]

Hi Chuck,

(Disclosure: I'm an RS SIG founding member.)

On Sat, Feb 10, 2018 at 7:14 PM, Chuck Davis <cjgun...@gmail.com> wrote:

> Hi James:
>
> Thanks for your response and the information in your (and other)
> blog(s).  I haven't had time to learn all the new features of jdk9 yet
> so a look at Flow was interesting.
>
> My first impression is that a time constraint would be better than
> destroying asynchrony.


The good new is that no asynchrony is destroyed. The first sentence of
reactive-streams.org quite literally says:

"Reactive Streams is an initiative to provide a standard for *asynchronous*
stream processing with non-blocking back pressure." (emphasis mine)


> Rather than telling the source to only send
> one message and thus effectively make it a synchronous process, tell
> the source to only send one message per 100 millis or 1000 millis or
> whatever is appropriate back-pressure while my other threads are doing
> their thing ??


Not only would that not work—as in solve the problem—and it would lead to
abysmal performance.



> When the WebSocket.Listener has a complete message
> trigger a notification event to registered listeners.  And if the
> programmer wants to process intermediate results that would not be too
> difficult for the listener to track.
>
> I can understand the potential need for "back-pressure" but I think
> conserving the asynchronous nature of WebSocket is a high priority as
> well.  Indeed, I've built my application on that feature of
> WebSockets.
>

Fortunately there is no tradeoff needed there. :-)


>
> Thanks again.  At least I'm a bit more enlightened about the issue
> being addressed.
>
> Chuck
>
>
>
>
>
> On Fri, Feb 9, 2018 at 11:38 PM, James Roper <ja...@lightbend.com> wrote:
> > Hi Chuck,
> >
> > Presumably this API is similar in intention to the request method used in
> > reactive streams (aka java.util.concurrent.Flow), that is, request is the
> > means by which backpressure is propagated. One major problem with JDK8
> > WebSockets is there's no way to asynchronously propagate backpressure,
> you
> > have to accept every message that comes as it comes, you can't tell the
> > other end to back off, which means if it's producing messages faster than
> > you can consume them, your only two options are to fail fast, or risk
> > running out of memory.  Reactive Streams solves this by requiring
> consumers
> > to signal demand for data/messages before they receive any - an
> invocation
> > of the request method is just saying how many more elements the consumer
> is
> > currently ready to receive, and can be invoked many times, as the
> consumer
> > processes messages and is ready to receive more. Generally, for Reactive
> > Streams, application developers are not expected to implement or invoke
> > these APIs directly, instead, they are expected to use reactive streams
> > implementations like Akka Streams, rxJava or Reactor, which efficiently
> > manage buffering and keeping the buffer at an appropriate level for the
> > application developer, so the application developer can just focus on
> their
> > business concerns.
> >
> > But that brings me to a problem that I'd like to give as feedback to the
> > implementers - this API is not Reactive Streams, and so therefore can't
> take
> > advantage of Reactive Streams implementations, and more problematic,
> can't
> > interop with other Reactive Streams sinks/sources. If I wanted to stream
> a
> > WebSocket into a message broker that supports Reactive Streams, I can't.
> I
> > would definitely hope that Reactive Streams support could be added to
> this
> > API, at a minimum as a wrapper, so that application developers can easily
> > focus on their business problems, plumbing and transforming messages from
> > one place to another, rather than having to deal with implementing
> > concurrent code to pass messages. It may well require wrapping messages
> in a
> > high level object - text, binary, ping, pong, etc, to differentiate
> between
> > the message types.
> >
> > For a broader context of how this would fit into the broader Reactive
> > Streams ecosystem, I've published a blog post of what it would look like
> if
> > Java EE/EE4J were to adopt Reactive Streams everywhere, as it happens
> this
> > also includes proposals for using Reactive Streams in the JSR 356
> WebSocket
> > spec:
> >
> > https://developer.lightbend.com/blog/2018-02-06-reactive-
> streams-ee4j/index.html
> >
> > Regards,
> >
> > James
> >
> > On 9 February 2018 at 19:16, Chuck Da

Re: websockets

[James Roper]

Hi Pavel,

I was wondering if you could address the concerns I raised about it not
actually using Reactive Streams. It would be great if, having created a
WebSocket, a user could just plumb it to another reactive streams
source/sink without writing any glue code, but with this, they're going to
have to write a Publisher/Subscriber first that wraps the current API. I
understand why the current API exists, it's because there's different types
of messages, binary, text, ping/pong, and I think it makes sense to keep
that API for simple use cases. But it would be nice if in addition, a
Publisher/Subscriber API was provided, otherwise application developers
using reactive streams are going to be reimplementing the same thing every
time they need it.

Regards,

James



On 11 February 2018 at 20:24, Pavel Rappo <pavel.ra...@oracle.com> wrote:

>
> > On 9 Feb 2018, at 18:16, Chuck Davis <cjgun...@gmail.com> wrote:
> >
> > I've been using jdk8 websockets to develop my desktop java
> > applications.  Now that jdk9 is on my machine I started looking at
> > websockets and I'm not at all sure I like what I see.  Can someone
> > familiar with this feature please explain the rationale for what is
> > happening?
> >
> > I'm concerned, at this initial stage, primarily by
> > WebSocket.request(long).  This "feature" seems to have at least two
> > very negative impacts:
> >
> > 1)  It appears to destroy the asynchronous nature of websockets;
> > 2)  It appears to place programmers in the impossible position of
> > guessing how many messages the server side might send.
> >
> > 1)  If everything has to stop while the client asks for more messages
> > asynchronous communication is nullified.  The jdk8 implementation is,
> > therefore, much more functional.
> >
> > 2)  It would appear the only logical use of WebSocket.request() would
> > be to use a long value of Long.MAX_VALUE since there is no way to know
> > how many messages may be received.  And what if the programmer asks
> > for 1 message and the next message has 3 parts.  We're screwed.
> > Additionally, the documentation specifically states that the
> > WebSocket.Listener does not distinguish between partial and whole
> > messages.  The jdk8 implementation decoders/encoders accumulate
> > messages and assemble them until the message is complete and then pass
> > it to the Endpoint -- much more satisfactory arrangement.
> >
> > I cannot fathom the meaning or improvement of this new wrinkle.
> >
> > Thanks for any insight
>
> Hello Chuck,
>
> James is right and this API is heavily influenced by the Reactive Streams
> [1].
> It is not about how many messages the peer is willing to send, it is about
> how
> many messages you are ready to receive. Because otherwise you leave things
> to
> chance. As without specific measures one should not expect a consumer and a
> producer to have the same throughput.
>
> Correct me if I'm wrong, but when you say that WebSocket has an
> asynchronous
> nature, I guess what you actually mean is that each peer can send data at
> will.
> A peer can be sending and receiving data simultaneously. This would rather
> be a
> bidirectional and full-duplex communication. It is the API that could be
> asynchronous. The API in question if fully asynchronous. A method that
> sends a
> message does not wait for the message to be sent, instead it returns a
> pending
> result of the operation. The same is true for requesting messages.
> WebSocket.request(long) acts asynchronously. It simply records how many
> extra
> messages the client is ready to receive. Finally, each of the listener's
> methods
> asks the user to signal when the user has processed the message it bears
> [2].
>
> The rationale behind this is to allow to avoid unnecessary copying of data
> between buffers when this is considered costly, and as usual with
> asynchrony,
> not to tie too many threads of execution.
>
> Consider the following example of a toy echoing client:
>
> WebSocket.Listener echo = new WebSocket.Listener() {
>
> @Override
> public void onOpen(WebSocket webSocket) {
> webSocket.request(1);
> }
>
> @Override
> public CompletionStage onBinary(WebSocket webSocket,
>ByteBuffer message,
>MessagePart part)
> {
> boolean isLast = (part == WHOLE || part == LAST);
> CompletableFuture sent =
> webSocket.sendBinary(message, isLast);
> sent.thenAcceptAsync(ws -> ws.request(1));
> return sent;
>

Re: websockets

[Pavel Rappo]

> On 9 Feb 2018, at 18:16, Chuck Davis <cjgun...@gmail.com> wrote:
> 
> I've been using jdk8 websockets to develop my desktop java
> applications.  Now that jdk9 is on my machine I started looking at
> websockets and I'm not at all sure I like what I see.  Can someone
> familiar with this feature please explain the rationale for what is
> happening?
> 
> I'm concerned, at this initial stage, primarily by
> WebSocket.request(long).  This "feature" seems to have at least two
> very negative impacts:
> 
> 1)  It appears to destroy the asynchronous nature of websockets;
> 2)  It appears to place programmers in the impossible position of
> guessing how many messages the server side might send.
> 
> 1)  If everything has to stop while the client asks for more messages
> asynchronous communication is nullified.  The jdk8 implementation is,
> therefore, much more functional.
> 
> 2)  It would appear the only logical use of WebSocket.request() would
> be to use a long value of Long.MAX_VALUE since there is no way to know
> how many messages may be received.  And what if the programmer asks
> for 1 message and the next message has 3 parts.  We're screwed.
> Additionally, the documentation specifically states that the
> WebSocket.Listener does not distinguish between partial and whole
> messages.  The jdk8 implementation decoders/encoders accumulate
> messages and assemble them until the message is complete and then pass
> it to the Endpoint -- much more satisfactory arrangement.
> 
> I cannot fathom the meaning or improvement of this new wrinkle.
> 
> Thanks for any insight

Hello Chuck,

James is right and this API is heavily influenced by the Reactive Streams [1].
It is not about how many messages the peer is willing to send, it is about how
many messages you are ready to receive. Because otherwise you leave things to
chance. As without specific measures one should not expect a consumer and a
producer to have the same throughput.

Correct me if I'm wrong, but when you say that WebSocket has an asynchronous
nature, I guess what you actually mean is that each peer can send data at will.
A peer can be sending and receiving data simultaneously. This would rather be a
bidirectional and full-duplex communication. It is the API that could be
asynchronous. The API in question if fully asynchronous. A method that sends a
message does not wait for the message to be sent, instead it returns a pending
result of the operation. The same is true for requesting messages.
WebSocket.request(long) acts asynchronously. It simply records how many extra
messages the client is ready to receive. Finally, each of the listener's methods
asks the user to signal when the user has processed the message it bears [2].

The rationale behind this is to allow to avoid unnecessary copying of data
between buffers when this is considered costly, and as usual with asynchrony,
not to tie too many threads of execution.

Consider the following example of a toy echoing client:

WebSocket.Listener echo = new WebSocket.Listener() {

@Override
public void onOpen(WebSocket webSocket) {
webSocket.request(1);
}

@Override
public CompletionStage onBinary(WebSocket webSocket,
   ByteBuffer message,
   MessagePart part)
{
boolean isLast = (part == WHOLE || part == LAST);
CompletableFuture sent = webSocket.sendBinary(message, 
isLast);
sent.thenAcceptAsync(ws -> ws.request(1));
return sent;
}
};

The next message will be requested after the previous one has been echoed back
to the server. In this example there is zero copying from the API user's
perspective.

Practically speaking, could it be any more asynchronous?

Mind you, going one-by-one (i.e. request(1) after consumption) is not the only
possibility. Back-pressure provides quite flexible control over consumption.
Have a look at java.util.concurrent.Flow. The spec for this type provides a very
nice example (SampleSubscriber) of buffering.

Chuck, if you still think this API is missing some crucial capability, I will
kindly ask you to provide an example of an application that is possible to
implement in a non-back-pressured model and is not possible in the proposed
WebSocket API.

Thanks,
-Pavel

---
[1] 
https://github.com/reactive-streams/reactive-streams-jvm/blob/3716ba57f17e55ec414aed7ccb1530251e193b61/README.md#goals-design-and-scope
[2] One does not have to signal the message has been processed in order to
receive the next message, if there are any. The signal is about handing over
the message contents, not about the back-pressure.

Re: websockets

[James Roper]

Hi Chuck,

Presumably this API is similar in intention to the request method used in
reactive streams (aka java.util.concurrent.Flow), that is, request is the
means by which backpressure is propagated. One major problem with JDK8
WebSockets is there's no way to asynchronously propagate backpressure, you
have to accept every message that comes as it comes, you can't tell the
other end to back off, which means if it's producing messages faster than
you can consume them, your only two options are to fail fast, or risk
running out of memory.  Reactive Streams solves this by requiring consumers
to signal demand for data/messages before they receive any - an invocation
of the request method is just saying how many more elements the consumer is
currently ready to receive, and can be invoked many times, as the consumer
processes messages and is ready to receive more. Generally, for Reactive
Streams, application developers are not expected to implement or invoke
these APIs directly, instead, they are expected to use reactive streams
implementations like Akka Streams, rxJava or Reactor, which efficiently
manage buffering and keeping the buffer at an appropriate level for the
application developer, so the application developer can just focus on their
business concerns.

But that brings me to a problem that I'd like to give as feedback to the
implementers - this API is not Reactive Streams, and so therefore can't
take advantage of Reactive Streams implementations, and more problematic,
can't interop with other Reactive Streams sinks/sources. If I wanted to
stream a WebSocket into a message broker that supports Reactive Streams, I
can't. I would definitely hope that Reactive Streams support could be added
to this API, at a minimum as a wrapper, so that application developers can
easily focus on their business problems, plumbing and transforming messages
from one place to another, rather than having to deal with implementing
concurrent code to pass messages. It may well require wrapping messages in
a high level object - text, binary, ping, pong, etc, to differentiate
between the message types.

For a broader context of how this would fit into the broader Reactive
Streams ecosystem, I've published a blog post of what it would look like if
Java EE/EE4J were to adopt Reactive Streams everywhere, as it happens this
also includes proposals for using Reactive Streams in the JSR 356 WebSocket
spec:

https://developer.lightbend.com/blog/2018-02-06-reactive-streams-ee4j/index.html

Regards,

James

On 9 February 2018 at 19:16, Chuck Davis <cjgun...@gmail.com> wrote:

> I've been using jdk8 websockets to develop my desktop java
> applications.  Now that jdk9 is on my machine I started looking at
> websockets and I'm not at all sure I like what I see.  Can someone
> familiar with this feature please explain the rationale for what is
> happening?
>
> I'm concerned, at this initial stage, primarily by
> WebSocket.request(long).  This "feature" seems to have at least two
> very negative impacts:
>
> 1)  It appears to destroy the asynchronous nature of websockets;
> 2)  It appears to place programmers in the impossible position of
> guessing how many messages the server side might send.
>
> 1)  If everything has to stop while the client asks for more messages
> asynchronous communication is nullified.  The jdk8 implementation is,
> therefore, much more functional.
>
> 2)  It would appear the only logical use of WebSocket.request() would
> be to use a long value of Long.MAX_VALUE since there is no way to know
> how many messages may be received.  And what if the programmer asks
> for 1 message and the next message has 3 parts.  We're screwed.
> Additionally, the documentation specifically states that the
> WebSocket.Listener does not distinguish between partial and whole
> messages.  The jdk8 implementation decoders/encoders accumulate
> messages and assemble them until the message is complete and then pass
> it to the Endpoint -- much more satisfactory arrangement.
>
> I cannot fathom the meaning or improvement of this new wrinkle.
>
> Thanks for any insight
>



-- 
*James Roper*
*Senior Octonaut*

Lightbend <https://www.lightbend.com/> – Build reactive apps!
Twitter: @jroper <https://twitter.com/jroper>

Re: websockets

[Chuck Davis]

I've been using jdk8 websockets to develop my desktop java
applications.  Now that jdk9 is on my machine I started looking at
websockets and I'm not at all sure I like what I see.  Can someone
familiar with this feature please explain the rationale for what is
happening?

I'm concerned, at this initial stage, primarily by
WebSocket.request(long).  This "feature" seems to have at least two
very negative impacts:

1)  It appears to destroy the asynchronous nature of websockets;
2)  It appears to place programmers in the impossible position of
guessing how many messages the server side might send.

1)  If everything has to stop while the client asks for more messages
asynchronous communication is nullified.  The jdk8 implementation is,
therefore, much more functional.

2)  It would appear the only logical use of WebSocket.request() would
be to use a long value of Long.MAX_VALUE since there is no way to know
how many messages may be received.  And what if the programmer asks
for 1 message and the next message has 3 parts.  We're screwed.
Additionally, the documentation specifically states that the
WebSocket.Listener does not distinguish between partial and whole
messages.  The jdk8 implementation decoders/encoders accumulate
messages and assemble them until the message is complete and then pass
it to the Endpoint -- much more satisfactory arrangement.

I cannot fathom the meaning or improvement of this new wrinkle.

Thanks for any insight