Issue #13043 - review of websocket Flushers for 12.1 #13250
Conversation
Signed-off-by: Lachlan Roberts <[email protected]>
Signed-off-by: Lachlan Roberts <[email protected]>
gregw
left a comment
gregw
left a comment
There was a problem hiding this comment.
Mostly looks good, but I think some comments are wrong with their upstream/downstream usage? which way the stream flows needs to be made clear in comments.
Then I'm not sure mixing in "Next" is a good idea. Isn't that just "Downstream"?
...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/WebSocketDemander.java
Outdated
Show resolved
Hide resolved
...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/WebSocketDemander.java
Show resolved
Hide resolved
...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/WebSocketDemander.java
Outdated
Show resolved
Hide resolved
...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/WebSocketDemander.java
Outdated
Show resolved
Hide resolved
Signed-off-by: Lachlan Roberts <[email protected]>
...t-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java
Outdated
Show resolved
Hide resolved
...t-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java
Outdated
Show resolved
Hide resolved
| boolean finished = deflate(callback); | ||
| _first = false; | ||
| // Provide the frames payload as input to the Deflater. | ||
| getDeflater().setInput(entry.getFrame().getPayload().slice()); |
There was a problem hiding this comment.
We do because currently the contract is that we do not consume the frame payload on a sendFrame (see #13290).
...mmon/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java
Outdated
Show resolved
Hide resolved
...cket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/FragmentingFlusher.java
Show resolved
Hide resolved
...ocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/WebSocketDemander.java
Show resolved
Hide resolved
| if (failure == null) | ||
| { | ||
| succeeded(); | ||
| } | ||
| else | ||
| { | ||
| callback.failed(failure); | ||
| failFlusher(failure); | ||
| } |
There was a problem hiding this comment.
This feels wrong, because if no failure, the whole IteratingCallback is succeeded, but not the callback parameter, and if there is a failure, the callback parameter is failed, but not the IteratingCallback (which is instead aborted).
There was a problem hiding this comment.
We can't succeed the callback parameter because we are not done with the frames payload, but if we have a non-null failure we need to fail it because it wasn't added to the state and so won't be failed later on.
I don't think we want to just fail the iterating callback. It is really a terminal failure so we want to ensure that the Demander is aborted. Looking at ICB.failed there are cases where it just turns into a NOOP, for example if it has already been completed.
There was a problem hiding this comment.
Please add a comment to the call to succeeded(), something like:
// Succeed this IteratingCallback, since we are not done with the frame's payload.
| * @param first true if this is the first time this entry is being processed. | ||
| * @return true to indicate that you have finished transforming this entry. | ||
| */ | ||
| protected abstract boolean onFrame(OutgoingEntry entry, boolean first); |
There was a problem hiding this comment.
Can the entry transformation be asynchronous, so that this should rather be:
| protected abstract boolean onFrame(OutgoingEntry entry, boolean first); | |
| protected abstract void onFrame(OutgoingEntry entry, boolean first, Promise<Boolean> promise); |
There was a problem hiding this comment.
I don't think there is a need for this.
And I don't see how this would work with the Flusher.process implementation.
| } | ||
| catch (Throwable x) | ||
| { | ||
| log.warn("Exception while notifying success of entry {}", entry, x); |
There was a problem hiding this comment.
Use LOG.info() like in the other cases of exceptions while notifying listeners.
| } | ||
| catch (Throwable x) | ||
| { | ||
| log.warn("Exception while notifying failure of entry {}", entry, x); |
There was a problem hiding this comment.
Debug or warn. info is for more expected stuff.
There was a problem hiding this comment.
Debug seems a bit light for this error.
Signed-off-by: Lachlan Roberts <[email protected]>
sbordet
left a comment
sbordet
left a comment
There was a problem hiding this comment.
LGTM, just the nit of adding a comment.
github-project-automation
bot
moved this from 👀 In review
to ✅ Done
in Jetty 12.1.0 - (FROZEN)
closes #13043
TransformingFlushertoWebSocketFlusherand update it to use anOutgoingEntryAPI instead offrame + callback + batch.onCompleteFailureonWebSocketFlusherWebSocketFlushernow wraps the callback needed to continue processing in theOutgoingEntryitself.DemandingFlushertoWebSocketDemander, and use a lock instead of multiple atomic variables for the implementation.