[WIP] Add SSE Support #1508
[WIP] Add SSE Support #1508
Conversation
There was a problem hiding this comment.
@aesteve thank you and sorry for the delay. I've made a first pass.
| import io.vertx.core.http.HttpClientOptions; | ||
| import io.vertx.ext.web.handler.sse.impl.EventSourceImpl; | ||
|
|
||
| public interface EventSource { |
vertx-web/src/main/java/io/vertx/ext/web/handler/sse/SSEHeaders.java
Outdated
Show resolved
Hide resolved
|
|
||
| @Test | ||
| public void noHeaderTextEventStreamHttpRequest() throws InterruptedException { | ||
| CountDownLatch latch = new CountDownLatch(1); |
There was a problem hiding this comment.
Formatting issues here (and other places). The project has an editorConfig file so if your IDE supports it, you should be able to reformat easily.
vertx-web/src/main/java/io/vertx/ext/web/handler/sse/SSEConnection.java
Outdated
Show resolved
Hide resolved
| SSEConnection retry(Long delay, String data); | ||
|
|
||
| @Fluent | ||
| SSEConnection data(List<String> data); |
There was a problem hiding this comment.
What's the use case for this?
There was a problem hiding this comment.
Sending multiple lines at once?
There was a problem hiding this comment.
I was wondering why, as a user, would you want to split content into multiple data events?
There was a problem hiding this comment.
🤷🏻♂️
In the docs here there's an example, but not really obvious:
data: Here's a system message of some kind that will get used
data: to accomplish some task.
We can remove it if you think it's not accurate. It can easily be achieved by invoking data() multiple times. No worries.
There was a problem hiding this comment.
I understood that you can send multiple data messages but I was wondering if there are cases when you need to split content.
There was a problem hiding this comment.
Actually, invoking data(...) multiple times doesn't really achieve the same thing.
Since invoking data(...) finishes the message with \n\n.
But in the example from the docs both data: are separated with just one carriage return, which means on the client side they'll be merged into one single String.
So this point is NOT addressed yet. We need to figure out how to mimic this example on the server, and fix the EventSource client so that it merges multiple data: lines into one single String to conform to the spec.
There was a problem hiding this comment.
There are different ways we can address this. One could be, in SSEConnection to have:
- a generic:
SSEConnection data(String msg, boolean endPacket)method - rewrite
SSEConnection data(String msg)method to bedefault SSEConnection data(String msg) { return data(msg, true); } - And then let the user invoke
data("...", false);data("......", false);data("last message");
What do you think of this @tsegismont if you agree, it'll allow me to test the EventSource counterpart (merging multi-line messages into a single String) as mentioned in the specs very easily, since the multi-line functionnality would be supported on the server.
There was a problem hiding this comment.
I implemented this, which allowed me to easily write a unit test for the EventSource client ability to merge multiline data into a Single String. Let me know what you think about it.
vertx-web/src/main/java/io/vertx/ext/web/handler/sse/SSEConnection.java
Outdated
Show resolved
Hide resolved
vertx-web/src/main/java/io/vertx/ext/web/handler/sse/SSEConnection.java
Outdated
Show resolved
Hide resolved
vertx-web/src/main/java/io/vertx/ext/web/handler/sse/SSEHandler.java
Outdated
Show resolved
Hide resolved
You should update the
The standard for Vert.x 4 is to have good old callback style as well as futurized versions of the async methods.
The compiler will tell you for every object annotated with
I haven't found anything odd except formatting and missing copyright headers. Other than that, I wonder if this shouldn't be split into two parts: the Vert.x Web handler in the Vert.x Web module, and the client in a separate module. What do you think @pmlopes ? |
|
Ok so let me create a check list or there's a chance I'll be missing a lot of stuff since there's a lot of work to do:
That's a ton of work to do, I'll give it a try by the end of february I think. Thanks a lot for all the reviews. |
…rward messages from the event bus to an SSE connection
…ulti-line messages in EventSource
|
Just waiting for a second review @tsegismont before next phase (javadocs + official docs). Plus, converting to draft until Doc is there, so that it doesn't get merged by accident; |
|
Note: about EventSource, would |
| } | ||
|
|
||
| @Fluent | ||
| EventSource connect(String path, Handler<AsyncResult<Void>> handler); |
| EventSource connect(String path, String lastEventId, Handler<AsyncResult<Void>> handler); | ||
|
|
||
| @Fluent | ||
| EventSource onMessage(Handler<String> messageHandler); |
| EventSource onMessage(Handler<String> messageHandler); | ||
|
|
||
| @Fluent | ||
| EventSource onEvent(String eventName, Handler<String> handler); |
| EventSource onEvent(String eventName, Handler<String> handler); | ||
|
|
||
| @Fluent | ||
| default EventSource onError(Handler<String> handler) { |
| } | ||
|
|
||
| @Fluent | ||
| default EventSource addEventListener(String eventName, Handler<String> handler) { |
| vertx.cancelTimer(retryTimerId); | ||
| retryTimerId = null; | ||
| } | ||
| if (client != null) { |
There was a problem hiding this comment.
move the client close outside of the synchronized block (after copying the reference to a local variable and setting the ref to null)
There was a problem hiding this comment.
To clarify, do you mean something like this?
@Override
public void close() {
HttpClient client;
synchronized (this) {
if (retryTimerId != null) {
vertx.cancelTimer(retryTimerId);
retryTimerId = null;
}
client = this.client;
this.client = null;
connected = false;
}
if (client != null) {
try {
client.close();
} catch (Exception e) {
log.error("An error occurred closing the EventSource: ", e);
}
}
}
| try { | ||
| client.close(); | ||
| } catch(Exception e ) { | ||
| e.printStackTrace(); |
| currentPacket = new SSEPacket(); | ||
| } | ||
| boolean terminated = currentPacket.append(buffer); | ||
| Optional<Handler<String>> eventHandler = Optional.empty(); |
There was a problem hiding this comment.
can we avoid using Optional here ?
|
|
||
| String lastId(); | ||
|
|
||
| @GenIgnore |
There was a problem hiding this comment.
GenIgnore is not needed I think
| SSEConnection comment(String comment); | ||
|
|
||
| @Fluent | ||
| SSEConnection retry(long delay); |
There was a problem hiding this comment.
I think we need to discuss the API, it seems to me, that this PR is trying to do client and server SSE on the vertx-web module (instead of being split between web, and web-client). It also looks (but i may be totally wrong here) that it tries to re-implement the sockjs bridge for SSE, which gets us to achieve the same in 2 different ways.
I think we should clearly define the goals, but we're already going on the right direction!
| } | ||
|
|
||
| @Fluent | ||
| SSEHandler connectHandler(Handler<SSEConnection> connection); |
There was a problem hiding this comment.
There should be some explanation why is this method needed/what it does.
| public synchronized void handle(RoutingContext context) { | ||
| HttpServerRequest request = context.request(); | ||
| HttpServerResponse response = context.response(); | ||
| List<String> acceptHeader = request.headers().getAll(HttpHeaders.ACCEPT.toString()); |
There was a problem hiding this comment.
Use the context.parsedHeaders() instead of handling the header yourself. This ensures order and quality are ensured. for example:
List<MIMEHeader> acceptableTypes = context.parsedHeaders().accept();
if(!acceptableTypes.isEmpty()) {
MIMEHeader selectedAccept = context.parsedHeaders()
.findBestUserAcceptedIn(acceptableTypes, "text/event-stream");
if (selectedAccept == null) {
return 406;
...| * This enumeration is public since it can be used as EventBus message headers, | ||
| * in case the user needs to forward messages from the event-bus with metadata (like "event:", or "id:") | ||
| */ | ||
| @VertxGen |
There was a problem hiding this comment.
I'm not sure we should have this. Having this means that we will now have 2 ways to achieve the same:
- the eventbus sockjs bridge
- the sse handler
IMHO, we should keep SSE as simple SSE. Instead have a single DataObject that defines the EventSource as per spec, or https://developer.mozilla.org/en-US/docs/Web/API/EventSource
{
event: String,
data: Buffer,
id: String,
retry: Integer
}
| import java.util.Map; | ||
| import java.util.Optional; | ||
|
|
||
| class SSEPacket { |
There was a problem hiding this comment.
This class should be combined with the SSEHeaders so we have a single DataObject to describe the Event.
| // A simple SSE handler that doesn't do anything specific apart accepting the connection | ||
| SSEHandler sseHandler = SSEHandler.create(); | ||
| sseHandler.connectHandler(connection -> { | ||
| this.connection = connection; // accept |
There was a problem hiding this comment.
Why do we need this (from an API perspective)? Access Control should be done at a previous handler IMHO (unless this isn't for access control)
| import io.vertx.core.json.JsonObject; | ||
|
|
||
| @DataObject(generateConverter = true) | ||
| public class EventSourceOptions extends HttpClientOptions { |
There was a problem hiding this comment.
I think this should go to the vertx-web-client module, not vertx-web
| import io.vertx.ext.web.RoutingContext; | ||
|
|
||
| @VertxGen | ||
| public interface SSEConnection { |
There was a problem hiding this comment.
Isn't this doing a similar task as SSEPack and SSEHeader?
|
Hey @aesteve, Have you had a chance to make any updates? I'd like to help out, but don't want to duplicate effort. Assuming not, @vietj / @pmlopes / @tsegismont, what's the best way to make this compatible with Vert.x 4.x? 1. 2. Given those two changes, I'm not sure how I would implement equivalent logic in I'll upload my attempt later today; any ideas/guidance would be appreciated. |
From: https://github.com/aesteve/vertx-sse
Original discussion: reactiverse/reactiverse#5
Status: Still WIP. Misses documentation + implementation must be discussed further on.
Motivation:
Provide a way to deal with Server-Sent-Events with Vert.x
SSE allows unidirectional communication between server and client in a pure HTTP way.
Status:
I migrated most of the code from my 3rd party implementation to make it officially supported by the Vert.x stack, as discussed in this reactiverse issue.
At this point, I'd like to receive feedback on:
Thank you.