Add WebSocketClient #4972

minwoox · 2023-06-20T14:38:22Z

Motivation
It would be nice if we also support WebSocket clients.

Modifications

Add SerializationFormat.WS for WebSocket.
Add WebSocketClient and its builder.
Extract the common part of HttpRequestSubciber to AbsractHttpRequestSubciber
and add WebSocketHttp1RequestSubscriber
Add ClientOpions.AUTO_FILL_ORIGIN_HEADER for adding the header automatically.
Add the pipeline Channel handler for WebSocket

Result

You can now send and receive WebSocket frames using WebSocketClient.

To-Do

Add WebSocketClientEventHandler
A lot of todos that are in this PR

core/src/main/java/com/linecorp/armeria/client/DefaultWebClient.java

core/src/main/java/com/linecorp/armeria/internal/common/DefaultSplitHttpResponse.java

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketClient.java

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketClientBuilder.java

ikhoon · 2023-07-04T09:28:50Z

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestSubscriber.java

+        RequestHeaders headers = request.headers();
+        final HttpSession session = HttpSession.get(channel());
+        final SerializationFormat serializationFormat = session.serializationFormat();
+        if (serializationFormat == SerializationFormat.WS) {


HttpSession is determined before starting subscription.We may add WebSocketHttp2RequestSubscriber and move this specialized logic there?

import com.linecorp.armeria.common.RequestHeaders; class WebSocketHttp2RequestSubscriber { RequestHeaders mapHeaders(RequestHeaders headers) { headers .toBuilder() .method(HttpMethod.CONNECT) ... } }

core/src/main/java/com/linecorp/armeria/client/ClientOptions.java

core/src/main/java/com/linecorp/armeria/client/Clients.java

ikhoon · 2023-07-04T09:44:26Z

core/src/main/java/com/linecorp/armeria/client/HttpSessionHandler.java

@@ -347,6 +359,10 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
            this.protocol = protocol;
            if (protocol == H1 || protocol == H1C) {
                final Http1ResponseDecoder responseDecoder = ctx.pipeline().get(Http1ResponseDecoder.class);
+                final boolean webSocket = serializationFormat == SerializationFormat.WS;


Q: Would it be possible to replace Http1ResponseDecoder with WebSocketResponseDecoder to modulize the WebSocket domain?

Split it. PTAL. 😉
a94f576

I thought we could completely replace Http1ResponseDecoder. The current implementation looks good.

This PR is already huge. If necessary, additional modularization could be taken in the future.

Changed to use WebSocketHttp1ClientChannelHandler instead. PTAL. 😉

core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestHandler.java

core/src/main/java/com/linecorp/armeria/client/AbstractWebClientBuilder.java

core/src/main/java/com/linecorp/armeria/client/Http1ResponseDecoder.java

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1RequestSubscriber.java

trustin · 2023-07-05T06:59:43Z

core/src/test/java/com/linecorp/armeria/client/websocket/WebSocketClientTest.java

+import com.linecorp.armeria.server.websocket.WebSocketServiceHandler;
+import com.linecorp.armeria.testing.junit5.server.ServerExtension;
+
+class WebSocketClientTest {


Can we also test against Jetty's WebSocket server? We could add it to :it:websocket.

You have caught my laziness. 🤣

core/src/test/java/com/linecorp/armeria/client/websocket/WebSocketClientHandshakeTest.java

core/src/main/java/com/linecorp/armeria/internal/common/websocket/WebSocketUtil.java

core/src/main/java/com/linecorp/armeria/internal/common/HttpHeadersUtil.java

core/src/main/java/com/linecorp/armeria/internal/common/DefaultSplitHttpResponse.java

core/src/main/java/com/linecorp/armeria/client/websocket/DefaultWebSocketClient.java

core/src/main/java/com/linecorp/armeria/internal/common/DefaultSplitHttpResponse.java

core/src/main/java/com/linecorp/armeria/client/websocket/DefaultWebSocketClient.java

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketSession.java

ikhoon · 2023-07-21T09:11:26Z

core/src/main/java/com/linecorp/armeria/client/HttpSessionHandler.java

@@ -347,6 +359,10 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
            this.protocol = protocol;
            if (protocol == H1 || protocol == H1C) {
                final Http1ResponseDecoder responseDecoder = ctx.pipeline().get(Http1ResponseDecoder.class);
+                final boolean webSocket = serializationFormat == SerializationFormat.WS;


I thought we could completely replace Http1ResponseDecoder. The current implementation looks good.

This PR is already huge. If necessary, additional modularization could be taken in the future.

ikhoon · 2023-07-21T09:26:33Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

+            return;
+        }
+        // Do not close the response right away to give a chance to WebSocketFrameDecoder to close the
+        // response normally if it receives a close frame.


I remember we had a chat about this behavior before. If we close the response instead of abortion, we don't need the delay, right?

I don't remember the conclusion of our discussion either 😅
Did we decide to close the connection from server-side? or did we decide to strictly follow the websocket rfc and close from client-side?

Changed to close the delegate directly. 😉

Did we decide to close the connection from server-side? or did we decide to strictly follow the websocket rfc and close from client-side?

Because we can't control and guarantee that, we just close the connection after close frames are sent and received in both sides. 😉

ikhoon · 2023-07-21T11:40:30Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

+        super(delegate, eventLoop, ctx, responseTimeoutMillis, maxContentLength);
+        final ClientRequestContextExtension extension = ctx.as(ClientRequestContextExtension.class);
+        if (extension != null) {
+            extension.setClosingResponseTask(cause -> {


Technically, this is a setter method that sets a Consumer to ctx.
For those usage, I think we may instead use attributes to store the Consumer.
What do you think?

I think in the future, we may also incorporate websocket ping/pong behavior to KeepAliveHandler which may require a more extensive callback handler.

At the moment since the callback isn't exposed, I'm fine with either way.

Changed to use attribute. BTW, I didn't change it for the server side because the original request might be used by others. Please, let me know if you think I need to change it as well. 😉

core/src/main/java/com/linecorp/armeria/common/stream/DeferredStreamMessage.java

core/src/main/java/com/linecorp/armeria/common/SerializationFormat.java

jrhee17

Mostly looks good! Got through roughly 2/3 😅 Will take a look at the rest tomorrow

jrhee17 · 2023-07-26T06:39:27Z

core/src/main/java/com/linecorp/armeria/client/WebSocketClientChannelHandler.java

+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        keepAliveHandler.onReadOrWrite();


Note: KeepAliveHandler at the moment won't support ping/pong

jrhee17 · 2023-07-26T08:33:42Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

+            return;
+        }
+        // Do not close the response right away to give a chance to WebSocketFrameDecoder to close the
+        // response normally if it receives a close frame.


I don't remember the conclusion of our discussion either 😅
Did we decide to close the connection from server-side? or did we decide to strictly follow the websocket rfc and close from client-side?

jrhee17 · 2023-07-26T13:41:47Z

core/src/main/java/com/linecorp/armeria/common/SessionProtocol.java

@@ -65,6 +65,9 @@ public enum SessionProtocol {

    private static final Set<SessionProtocol> HTTPS_VALUES = Sets.immutableEnumSet(HTTPS, H1, H2);

+    private static final Set<SessionProtocol> HTTP_AND_HTTPS_VALUES =
+            Sets.immutableEnumSet(HTTPS, HTTP, H1, H1C, H2, H2C);


Suggested change

Sets.immutableEnumSet(HTTPS, HTTP, H1, H1C, H2, H2C);

Sets.union(HTTP_VALUES, HTTPS_VALUES);

Sets.union has unnecessary operations inside its method. For example, it checks if each set contains the element from the other set in size() method:

public int size() { int size = set1.size(); for (E e : set2) { if (!set1.contains(e)) { size++; } } return size; }

So let me just use the current logic. 😉

jrhee17 · 2023-07-26T13:48:55Z

core/src/main/java/com/linecorp/armeria/common/Scheme.java

+                    } else if (SessionProtocol.HTTPS == p) {
+                        schemes.put("wss", scheme);
+                    }
+                }


Question) Why can't this be outside of the double for loop?

schemes.put("ws", new Scheme(SerializationFormat.WS, SessionProtocol.HTTP)); schemes.put("wss", new Scheme(SerializationFormat.WS, SessionProtocol.HTTPS));

That will create two new Scheme(SerializationFormat.WS, SessionProtocol.HTTP) instances whose references are different and it wouldn't work if we use == instead of scheme.equals(others).

jrhee17 · 2023-07-26T14:22:19Z

core/src/main/java/com/linecorp/armeria/internal/common/HttpHeadersUtil.java

+            if (internalHeaders.isEmpty()) {
+                return headers;
+            }
+            if (internalHeaders.size() == 1 && internalHeaders.contains(HttpHeaderNames.USER_AGENT) &&


This looks like extra overhead since I think most requests will contain

internalHeaders.authority

internalHeaders.scheme

What was the motivation of this change?

That is correct! 👍 Let me just revert it and leave the following if condition. 😉

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketClientHandshakeException.java

jrhee17 · 2023-07-26T14:45:32Z

core/src/main/java/com/linecorp/armeria/client/websocket/DefaultWebSocketClient.java

+
+final class DefaultWebSocketClient implements WebSocketClient {
+
+    static final WebSocketClient DEFAULT = new DefaultWebSocketClient();


Why can't we just use the builder?

Suggested change

static final WebSocketClient DEFAULT = new DefaultWebSocketClient();

static final WebSocketClient DEFAULT = new WebSocketClientBuilder(UNDEFINED_URI).build()

Should've pushed my change before you left this comment. 🤣

jrhee17 · 2023-07-26T15:02:21Z

core/src/main/java/com/linecorp/armeria/internal/client/DefaultClientRequestContext.java

@@ -438,7 +440,7 @@ private void failEarly(Throwable cause) {
    }

    // TODO(ikhoon): Consider moving the logic for filling authority to `HttpClientDelegate.exceute()`.
-    private void autoFillSchemeAndAuthority() {
+    private void autoFillSchemeAuthorityAndOrigin() {


While double checking, it seems like the rfc specifies origin as optional.

Optionally, an |Origin| header field. This header field is sent
by all browser clients. A connection attempt lacking this
header field SHOULD NOT be interpreted as coming from a browser
client.

I'm fine with this change if this is important for compatibility with other server implementations, but I wanted to double-check if we want to add this header by default.

ref: https://datatracker.ietf.org/doc/html/rfc6455#section-4.2.1

Because we have decided to reject a WebSocket request if the origin header is missing from the WebSocketService, I think we should send the origin header by default for the client. 😉

jrhee17

Left a couple of comments but mostly looks good 👍

core/src/main/java/com/linecorp/armeria/common/stream/DeferredStreamMessage.java

jrhee17 · 2023-07-27T02:54:20Z

core/src/main/java/com/linecorp/armeria/client/websocket/DefaultWebSocketClient.java

+                });
+
+        final CompletableFuture<WebSocketSession> result = new CompletableFuture<>();
+        split.headers().handle((responseHeaders, cause) -> {


Question) If endpoint selection fails and HttpResponse.ofFailure(cause) is returned, is it possible that headersFuture is never triggered and thus result is never completed?

I actually think even actualSessionProtocol(ctx) can fail if ctx is never created.

armeria/core/src/main/java/com/linecorp/armeria/client/DefaultWebClient.java

Line 99 in 548793d

return abortRequestAndReturnFailureResponse(req, new IllegalArgumentException(

If HttpResponse.ofFailure(cause) is returned, then

SplitHttpMessageSubscriber.onError() is called:

armeria/core/src/main/java/com/linecorp/armeria/internal/common/SplitHttpMessageSubscriber.java

Line 259 in f0818b9

doOnCompletion(cause);

DefaultSplitHttpResponse.doOnCompletion is called:

armeria/core/src/main/java/com/linecorp/armeria/internal/common/DefaultSplitHttpResponse.java

Line 102 in 9301b1c

protected void doOnCompletion(@Nullable Throwable cause) {

Thus, headersFuture is complete exceptionally. Did you mean something else?

I actually think even actualSessionProtocol(ctx) can fail if ctx is never created.

actualSessionProtocol(ctx) is invoked only when the client receives the ResponseHeaders from the server. if ctx is never created and HttpResponse.ofFailure(cause); is returned, split.headers().handle((responseHeaders, cause) -> {...} will be called directly with the cause.

jrhee17 · 2023-07-27T03:43:49Z

core/src/main/java/com/linecorp/armeria/internal/client/DefaultClientRequestContext.java

+                            new DefaultClientRequestContext(this, id, req, rpcReq, endpoint, null,
+                                                            sessionProtocol(), newHeaders.method(), reqTarget);
+                    // TODO(minwoox): Consider adding serizalizationFormat to ClientRequestContext constructor.
+                    ctx.logBuilder().serializationFormat(log().partial().serializationFormat());


Isn't this already being done?

armeria/core/src/main/java/com/linecorp/armeria/internal/client/ClientUtil.java

Line 189 in 548793d

logBuilder.serializationFormat(partial.serializationFormat());

Oops, let me revert it. 😓 Thanks!

core/src/main/java/com/linecorp/armeria/client/websocket/DefaultWebSocketClient.java

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketClient.java

jrhee17 · 2023-07-27T05:28:17Z

core/src/main/java/com/linecorp/armeria/internal/client/HttpSession.java

+    SerializationFormat serializationFormat();
+
+    /**
+     * Returns the explicit {@link SessionProtocol} of this {@link HttpSession} uses.


Suggested change

* Returns the explicit {@link SessionProtocol} of this {@link HttpSession} uses.

* Returns the explicit {@link SessionProtocol} of this {@link HttpSession}.

jrhee17 · 2023-07-27T05:29:58Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

+        super(delegate, eventLoop, ctx, responseTimeoutMillis, maxContentLength);
+        final ClientRequestContextExtension extension = ctx.as(ClientRequestContextExtension.class);
+        if (extension != null) {
+            extension.setClosingResponseTask(cause -> {


I think in the future, we may also incorporate websocket ping/pong behavior to KeepAliveHandler which may require a more extensive callback handler.

At the moment since the callback isn't exposed, I'm fine with either way.

jrhee17 · 2023-07-27T05:32:23Z

core/src/main/java/com/linecorp/armeria/internal/client/ClientRequestContextExtension.java

+    /**
+     * Sets the {@link Consumer} which will be invoked when the response is closed.
+     */
+    void setClosingResponseTask(Consumer<@Nullable Throwable> closingResponseTask);


What do you think of adding the word websocket in the naming so that we don't have to look up what this does every time?

i.e. webSocketResponseCloseTask or onWebSocketCloseFrame

jrhee17 · 2023-07-27T05:42:45Z

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketSession.java

+    /**
+     * Sets the {@link WebSocket} which is used to send WebSocket frames to the server.
+     */
+    public boolean setOutbound(Publisher<? extends WebSocketFrame> outbound) {


Do we use set in our setters instead of the record style?

Suggested change

public boolean setOutbound(Publisher<? extends WebSocketFrame> outbound) {

public boolean outbound(Publisher<? extends WebSocketFrame> outbound) {

Also, do you think it would be confusing to add a shortcut in case users just want to write directly?

val writer = WebsocketSession.outbound() // creates a writer internally and sets to `outboundFuture` writer.write("hi")

Do we use set in our setters instead of the record style?

I believe so and this is one of examples:

armeria/core/src/main/java/com/linecorp/armeria/client/ClientRequestContext.java

Lines 307 to 323 in 7da39b8

/**

* Returns the amount of time allowed until the initial write attempt of the current {@link Request}

* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.

*/

long writeTimeoutMillis();

/**

* Sets the amount of time allowed until the initial write attempt of the current {@link Request}

* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.

*/

void setWriteTimeoutMillis(long writeTimeoutMillis);

/**

* Sets the amount of time allowed until the initial write attempt of the current {@link Request}

* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.

*/

void setWriteTimeout(Duration writeTimeout);

😉

, do you think it would be confusing to add a shortcut in case users just want to write directly?

That's a good suggestion. 😉

jrhee17 · 2023-07-27T05:43:34Z

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketSession.java

+
+/**
+ * A WebSocket session that is created after {@link WebSocketClient#connect(String)} succeeds.
+ * You can start sending {@link WebSocketFrame}s via {@link #setOutbound(WebSocket)}. You can also subscribe to


Suggested change

* You can start sending {@link WebSocketFrame}s via {@link #setOutbound(WebSocket)}. You can also subscribe to

* You can start sending {@link WebSocketFrame}s via {@link #setOutbound(Publisher)}. You can also subscribe to

jrhee17 · 2023-07-27T05:59:40Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ResponseWrapper.java

+            return;
+        }
+        // Do not close the response right away to give a chance to WebSocketFrameDecoder to close the
+        // response normally if it receives a close frame.


I've been staring at this piece of code for 30min just to remember why ClosedSessionException was excluded 😅

Suggested change

// response normally if it receives a close frame.

// response normally if it receives a close frame instead of completing

// exceptionally due to a ClosedSessionException.

codecov · 2023-08-10T08:55:47Z

Codecov Report

Patch coverage: 70.18% and project coverage change: -0.10% ⚠️

Comparison is base (1ce4c69) 74.30% compared to head (ec847a4) 74.21%.
Report is 4 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #4972      +/-   ##
============================================
- Coverage     74.30%   74.21%   -0.10%     
- Complexity    19600    19801     +201     
============================================
  Files          1682     1698      +16     
  Lines         72260    73091     +831     
  Branches       9242     9357     +115     
============================================
+ Hits          53695    54241     +546     
- Misses        14222    14417     +195     
- Partials       4343     4433      +90

Files Changed	Coverage Δ
...ia/client/BlockingWebClientRequestPreparation.java	`38.61% <0.00%> (-0.79%)`	⬇️
...a/client/FutureTransformingRequestPreparation.java	`29.68% <0.00%> (-0.48%)`	⬇️
.../linecorp/armeria/client/Http2ResponseDecoder.java	`72.46% <ø> (ø)`
...m/linecorp/armeria/client/HttpResponseDecoder.java	`0.00% <ø> (-79.42%)`	⬇️
...linecorp/armeria/client/RestClientPreparation.java	`38.63% <0.00%> (-0.90%)`	⬇️
...armeria/client/TransformingRequestPreparation.java	`16.16% <0.00%> (-0.34%)`	⬇️
...rp/armeria/client/WebClientRequestPreparation.java	`72.47% <0.00%> (-0.68%)`	⬇️
...t/websocket/WebSocketClientHandshakeException.java	`0.00% <0.00%> (ø)`
...orp/armeria/common/AbstractHttpMessageBuilder.java	`73.68% <0.00%> (-5.57%)`	⬇️
...om/linecorp/armeria/common/HttpRequestBuilder.java	`87.50% <0.00%> (-2.83%)`	⬇️
... and 51 more

... and 29 files with indirect coverage changes

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

minwoox · 2023-08-14T07:26:06Z

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestSubscriber.java

+
+import io.netty.channel.Channel;
+
+abstract class AbstractHttpRequestSubscriber extends AbstractHttpRequestHandler


The logic in this class isn't changed.
HttpRequestSubscriber is renamed to this AbstractHttpRequestSubscriber class and the new HttpRequestSubscriber extends this class.

minwoox · 2023-08-14T07:28:05Z

core/src/main/java/com/linecorp/armeria/client/AbstractHttpResponseDecoder.java

+import io.netty.util.collection.IntObjectHashMap;
+import io.netty.util.collection.IntObjectMap;
+
+abstract class AbstractHttpResponseDecoder implements HttpResponseDecoder {


HttpResponseDecoder now becomes AbstractHttpResponseDecoder and the new interface HttpResponseDecoder is added for WebSocketHttp1ClientChannelHandler.

minwoox · 2023-08-14T07:29:04Z

core/src/main/java/com/linecorp/armeria/client/Http1ResponseDecoder.java

        super(channel, InboundTrafficController.ofHttp1(channel));
+        final long idleTimeoutMillis = clientFactory.idleTimeoutMillis();


Moved the logic for creating keepAliveHandler into this constructor and keepAliveHandler is not nullable anymore.

minwoox · 2023-08-14T07:30:43Z

core/src/main/java/com/linecorp/armeria/client/HttpClientDelegate.java

-            pool.acquireLater(protocol, key, timingsBuilder).handle((newPooledChannel, cause) -> {
-                logSession(ctx, newPooledChannel, timingsBuilder.build());
-                if (cause == null) {
-                    doExecute(newPooledChannel, ctx, req, res);
-                } else {
-                    final UnprocessedRequestException wrapped = UnprocessedRequestException.of(cause);
-                    handleEarlyRequestException(ctx, req, wrapped);
-                    res.close(wrapped);
-                }
-                return null;
-            });
+            pool.acquireLater(protocol, serializationFormat, key, timingsBuilder)
+                .handle((newPooledChannel, cause) -> {
+                    logSession(ctx, newPooledChannel, timingsBuilder.build());
+                    if (cause == null) {
+                        doExecute(newPooledChannel, ctx, req, res);
+                    } else {
+                        final UnprocessedRequestException wrapped = UnprocessedRequestException.of(cause);
+                        handleEarlyRequestException(ctx, req, wrapped);
+                        res.close(wrapped);
+                    }
+                    return null;
+                });


Adding serializationFormat to acquireLater method is the only change.

minwoox · 2023-08-14T07:32:26Z

core/src/main/java/com/linecorp/armeria/client/HttpResponseWrapper.java

+import io.netty.channel.EventLoop;
+import io.netty.util.concurrent.EventExecutor;
+
+class HttpResponseWrapper implements StreamWriter<HttpObject> {


This class was in HttpResponseDecoder.
close method is a bit changed to support WebSocket.

minwoox · 2023-08-14T07:33:14Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.util.ReferenceCountUtil;
+
+final class WebSocketHttp1ClientChannelHandler extends ChannelDuplexHandler implements HttpResponseDecoder {


This class is used instead of Http1ResponseDecoder for WebSocket.

Optional) WebSocketHttp1ClientChannelHandler could be split into WebSocketHttp1UpgradleHandler and WebSocketHttp1ResponseHandler to abide by SRP in future work.

I thought it's okay in terms of handling the series of WebSocket responses. Will consider splitting it once the logic becomes complicated. 😉

core/src/main/java/com/linecorp/armeria/client/AbstractHttpRequestSubscriber.java

core/src/main/java/com/linecorp/armeria/client/AbstractWebClientBuilder.java

core/src/main/java/com/linecorp/armeria/client/Http1ResponseDecoder.java

core/src/main/java/com/linecorp/armeria/client/HttpChannelPool.java

core/src/main/java/com/linecorp/armeria/client/HttpSessionHandler.java

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

core/src/main/java/com/linecorp/armeria/common/SessionProtocol.java

core/src/test/java/com/linecorp/armeria/client/Http1ResponseDecoderTest.java

github-actions · 2023-08-21T03:41:53Z

🔍 Build Scan® (commit: `ec847a4`)

Job name	Status	Build Scan®
build-windows-latest-jdk-19	✅	https://ge.armeria.dev/s/5zoxi63n5vyza
build-self-hosted-jdk-8	✅	https://ge.armeria.dev/s/zxfau6snbdyjm
build-self-hosted-jdk-19-snapshot-blockhound	✅	https://ge.armeria.dev/s/wqjhxrjckbchu
build-self-hosted-jdk-17-min-java-17-coverage	✅	https://ge.armeria.dev/s/gzjqmd6ipayec
build-self-hosted-jdk-17-min-java-11	✅	https://ge.armeria.dev/s/4kbdiuz7glpte
build-self-hosted-jdk-17-leak	✅	https://ge.armeria.dev/s/itbj3qv2k5x7i
build-self-hosted-jdk-11	✅	https://ge.armeria.dev/s/7xb3mknh5ueas
build-macos-12-jdk-19	✅	https://ge.armeria.dev/s/4yi4zbazrpa7a

trustin

Left two nits but:

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

trustin · 2023-08-21T04:24:29Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

+                        }
+                        if (!res.tryWriteData(HttpData.wrap(data.retain()))) {
+                            ctx.close();
+                        }
                    }


Missing break?

Fixed. 😉

ikhoon · 2023-08-18T02:15:07Z

core/src/main/java/com/linecorp/armeria/client/HttpSessionHandler.java

+                if (serializationFormat == SerializationFormat.WS) {
+                    responseDecoder = ctx.pipeline().get(WebSocketHttp1ClientChannelHandler.class);
+                    keepAliveHandler = responseDecoder.keepAliveHandler();
                } else {
-                    keepAliveHandler = new NoopKeepAliveHandler();
+                    final Http1ResponseDecoder http1ResponseDecoder =
+                            ctx.pipeline().get(Http1ResponseDecoder.class);
+                    http1ResponseDecoder.maybeInitializeKeepAliveHandler(ctx);
+                    responseDecoder = http1ResponseDecoder;
+                    keepAliveHandler = http1ResponseDecoder.keepAliveHandler();


Q) Could we generalize this? I see the only difference is maybeInitializeKeepAliveHandler(ctx) which might do nothing in WebSocketHttp1ClientChannelHandler.

final HttpResponseDecoder responseDecoder = ctx.pipeline().get(HttpResponseDecoder.class); // Noop for WebSocketHttp1ClientChannelHandler responseDecoder.maybeInitializeKeepAliveHandler(ctx); keepAliveHandler = responseDecoder.keepAliveHandler();

Changed to call keepAliveHandler.initialize(ctx); directly. 😉

ikhoon · 2023-08-18T02:21:22Z

core/src/main/java/com/linecorp/armeria/client/HttpSessionHandler.java

+            final AbstractHttpRequestSubscriber subscriber;
+            if (serializationFormat == SerializationFormat.WS) {
+                if (protocol.isExplicitHttp1()) {
+                    subscriber = new WebSocketHttp1RequestSubscriber(
+                            channel, requestEncoder, responseDecoder, req, res, ctx, writeTimeoutMillis);
+                } else {
+                    assert protocol.isExplicitHttp2();
+                    subscriber = new WebSocketHttp2RequestSubscriber(
+                            channel, requestEncoder, responseDecoder, req, res, ctx, writeTimeoutMillis);
+                }
+            } else {
+                subscriber = new HttpRequestSubscriber(
+                        channel, requestEncoder, responseDecoder, req, res, ctx, writeTimeoutMillis);
            }


nit: What do you think of adding a factory method to AbstractHttpRequestSubscriber to encapsulate the dispatch?

AbstractHttpRequestSubscriber subscriber = AbstractHttpRequestSubscriber.of(channel, requestEncoder, responseDecoder, req, res, ctx, writeTimeoutMillis);

Nice abstraction. 👍

ikhoon · 2023-08-18T02:32:36Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

+            final StringBuilder buf = tempThreadLocals.stringBuilder();
+            buf.append("unexpected message type: " + msg.getClass().getName() +
+                       " (expected: " + HttpResponse.class.getName() + ", channel: " + ctx.channel() + ')');
+            fail(ctx, new ProtocolViolationException(buf.toString()));


Should we call fail() after the try-with-resource block because res.close(cause) in fail() can invoke callbacks? It is difficult to statically analyze what will be called.

That's a good point. 👍
Also, fix it in Http1ResponseDecoder.

ikhoon · 2023-08-18T02:42:43Z

core/src/main/java/com/linecorp/armeria/common/SerializationFormat.java

@@ -91,7 +104,7 @@ public final class SerializationFormat implements Comparable<SerializationFormat
    private static SerializationFormat register(
            BiMap<String, SerializationFormat> uriTextToFormats,
            Multimap<MediaType, SerializationFormat> simplifiedMediaTypeToFormats,
-            SerializationFormatProvider.Entry entry) {


ikhoon · 2023-08-21T05:37:26Z

core/src/main/java/com/linecorp/armeria/client/AbstractHttpResponseDecoder.java

+        return !session().isAcquirable() && !hasUnfinishedResponses();
+    }
+
+    static Exception contentTooLargeException(HttpResponseWrapper res, long transferred) {


nit:

Suggested change

static Exception contentTooLargeException(HttpResponseWrapper res, long transferred) {

static ContentTooLargeException contentTooLargeException(HttpResponseWrapper res,

long transferred) {

ikhoon · 2023-08-21T05:45:46Z

core/src/main/java/com/linecorp/armeria/client/HttpResponseWrapper.java

+        if (requestAutoAbortDelayMillis == 0) {
+            request.abort(ResponseCompleteException.get());
+            return;
+        }
+        if (requestAutoAbortDelayMillis > 0 &&
+            requestAutoAbortDelayMillis < Long.MAX_VALUE) {


nit: Would be clear to use a single if...else chain since it only sees requestAutoAbortDelayMillis?

It doesn't do anything when the value is negative or MAX_VALUE. Let me use the early-return style. 😉

ikhoon · 2023-08-21T05:52:25Z

core/src/main/java/com/linecorp/armeria/client/WebSocketHttp1ClientChannelHandler.java

+import io.netty.handler.codec.http.HttpUtil;
+import io.netty.util.ReferenceCountUtil;
+
+final class WebSocketHttp1ClientChannelHandler extends ChannelDuplexHandler implements HttpResponseDecoder {


Optional) WebSocketHttp1ClientChannelHandler could be split into WebSocketHttp1UpgradleHandler and WebSocketHttp1ResponseHandler to abide by SRP in future work.

ikhoon · 2023-08-21T06:00:16Z

core/src/main/java/com/linecorp/armeria/internal/common/HttpHeadersUtil.java

@@ -92,6 +92,19 @@ public static RequestHeaders mergeRequestHeaders(RequestHeaders headers,
            headers.contains(HttpHeaderNames.USER_AGENT)) {
            return headers;
        }
+        if (defaultHeaders.isEmpty() && additionalHeaders.isEmpty()) {
+            boolean containInternalHeaders = true;


nit:

Suggested change

boolean containInternalHeaders = true;

boolean containsAllInternalHeaders = true;

ikhoon

Thanks, @minwoox 🚀🚀

ikhoon · 2023-08-21T08:06:04Z

core/src/main/java/com/linecorp/armeria/client/websocket/WebSocketSession.java

+    public void setOutbound(Publisher<? extends WebSocketFrame> outbound) {
+        requireNonNull(outbound, "outbound");
+        if (outboundFuture.isDone()) {
+            throw new IllegalStateException("outbound() or setOutbound() has been already called.");


Should we call outbound.abort() if outbound is an instance of StreamMessage?

Yeah, we need it. Thanks! 😉

jrhee17

👍 👍 👍

[WIP] Add WebSocketClient

00e9d6d

minwoox requested review from ikhoon, jrhee17 and trustin as code owners June 20, 2023 14:38

minwoox added the new feature label Jun 20, 2023

minwoox added this to the 1.25.0 milestone Jun 20, 2023

minwoox added 4 commits June 29, 2023 20:39

WIP

d1b2fa9

Redesign API

9299aa9

Merge branch 'main' into WebSocketClient

2727bba

Add more tests

d832e8b

minwoox changed the title ~~[WIP] Add WebSocketClient~~ Add WebSocketClient Jun 30, 2023

minwoox mentioned this pull request Jun 30, 2023

WebSocket client #4234

Closed

minwoox added 3 commits June 30, 2023 15:30

Add Javadoc

dddfeeb

Add requiresNewConnection

bb6975a

Add WebSocketSession.abort()

eddb125

ikhoon reviewed Jul 4, 2023

View reviewed changes

trustin reviewed Jul 5, 2023

View reviewed changes

minwoox added 3 commits July 7, 2023 23:28

Address comments from @ikhoon and @trustin

9301b1c

Address more

223b86a

Fix

35ade23

minwoox mentioned this pull request Jul 11, 2023

Remove contentType from HttpMessageSetters.content() methods #5020

Open

Add WebSocketHttp1ResponseWrapper

a94f576

ikhoon reviewed Jul 21, 2023

View reviewed changes

minwoox added 2 commits July 25, 2023 13:04

Merge branch 'main' into WebSocketClient

075db60

Merge branch 'main' into WebSocketClient

a0cc9a5

jrhee17 reviewed Jul 26, 2023

View reviewed changes

jrhee17 reviewed Jul 27, 2023

View reviewed changes

Parially address comments from @ikhoon

4a21b49

minwoox added 4 commits August 10, 2023 17:56

Merge branch 'main' into WebSocketClient

7b63523

Parially address comments from @jrhee17

63f625c

Use WebSocketHttp1ClientChannelHandler instead of Http1ResponseDecoder

d180dfd

Fix

1dcb57c

minwoox commented Aug 14, 2023

View reviewed changes

minwoox added 4 commits August 14, 2023 17:05

Fix test

86fcf40

Add unstable api

4fdefc8

Remove unused method

529961c

Merge branch 'main' into WebSocketClient

cefafba

trustin reviewed Aug 18, 2023

View reviewed changes

minwoox added 2 commits August 21, 2023 12:07

Address comments from @trustin

e0b1f98

Merge branch 'main' into WebSocketClient

2c690e8

minwoox force-pushed the WebSocketClient branch from a0684e3 to 2c690e8 Compare August 21, 2023 03:09

minwoox added 2 commits August 21, 2023 12:34

Add comment

81b3670

Fix lint

9b6ff8b

trustin approved these changes Aug 21, 2023

View reviewed changes

Add breaks

9a5ff19

ikhoon reviewed Aug 21, 2023

View reviewed changes

Address comments from @ikhoon

51a3a3c

ikhoon approved these changes Aug 21, 2023

View reviewed changes

Abort Publisher

0e8f1bf

minwoox mentioned this pull request Aug 22, 2023

Refactor Bootstrap retrieval in HttpChannelPool #5129

Closed

Fix flaky

ec847a4

jrhee17 approved these changes Aug 22, 2023

View reviewed changes

jrhee17 merged commit dc2bdc6 into line:main Aug 22, 2023

minwoox deleted the WebSocketClient branch August 22, 2023 05:30

	Sets.immutableEnumSet(HTTPS, HTTP, H1, H1C, H2, H2C);
	Sets.union(HTTP_VALUES, HTTPS_VALUES);


		final class DefaultWebSocketClient implements WebSocketClient {

		static final WebSocketClient DEFAULT = new DefaultWebSocketClient();

	* Returns the explicit {@link SessionProtocol} of this {@link HttpSession} uses.
	* Returns the explicit {@link SessionProtocol} of this {@link HttpSession}.

	public boolean setOutbound(Publisher<? extends WebSocketFrame> outbound) {
	public boolean outbound(Publisher<? extends WebSocketFrame> outbound) {

	/**
	* Returns the amount of time allowed until the initial write attempt of the current {@link Request}
	* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.
	*/
	long writeTimeoutMillis();

	/**
	* Sets the amount of time allowed until the initial write attempt of the current {@link Request}
	* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.
	*/
	void setWriteTimeoutMillis(long writeTimeoutMillis);

	/**
	* Sets the amount of time allowed until the initial write attempt of the current {@link Request}
	* succeeds. This value is initially set from {@link ClientOptions#WRITE_TIMEOUT_MILLIS}.
	*/
	void setWriteTimeout(Duration writeTimeout);

	* You can start sending {@link WebSocketFrame}s via {@link #setOutbound(WebSocket)}. You can also subscribe to
	* You can start sending {@link WebSocketFrame}s via {@link #setOutbound(Publisher)}. You can also subscribe to

	// response normally if it receives a close frame.
	// response normally if it receives a close frame instead of completing
	// exceptionally due to a ClosedSessionException.


		import io.netty.channel.Channel;

		abstract class AbstractHttpRequestSubscriber extends AbstractHttpRequestHandler

		super(channel, InboundTrafficController.ofHttp1(channel));
		final long idleTimeoutMillis = clientFactory.idleTimeoutMillis();

	static Exception contentTooLargeException(HttpResponseWrapper res, long transferred) {
	static ContentTooLargeException contentTooLargeException(HttpResponseWrapper res,
	long transferred) {

	boolean containInternalHeaders = true;
	boolean containsAllInternalHeaders = true;

Add WebSocketClient #4972

Add WebSocketClient #4972

Conversation

minwoox commented Jun 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrhee17 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

minwoox Aug 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

minwoox Aug 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

minwoox Aug 10, 2023 • edited Loading

Choose a reason for hiding this comment

jrhee17 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

minwoox Aug 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Aug 10, 2023 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Aug 21, 2023 • edited Loading

🔍 Build Scan® (commit: ec847a4)

trustin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ikhoon left a comment

Choose a reason for hiding this comment

minwoox commented Jun 20, 2023 •

edited

Loading

minwoox Aug 10, 2023 •

edited

Loading

minwoox Aug 10, 2023 •

edited

Loading

minwoox Aug 10, 2023 •

edited

Loading

minwoox Aug 10, 2023 •

edited

Loading

codecov bot commented Aug 10, 2023 •

edited

Loading

github-actions bot commented Aug 21, 2023 •

edited

Loading

🔍 Build Scan® (commit: `ec847a4`)