diff --git a/framework-docs/modules/ROOT/pages/web/websocket/fallback.adoc b/framework-docs/modules/ROOT/pages/web/websocket/fallback.adoc index bdbac08ae32..5c39cc7ff48 100644 --- a/framework-docs/modules/ROOT/pages/web/websocket/fallback.adoc +++ b/framework-docs/modules/ROOT/pages/web/websocket/fallback.adoc @@ -257,7 +257,7 @@ An `XhrTransport`, by definition, supports both `xhr-streaming` and `xhr-polling from a client perspective, there is no difference other than in the URL used to connect to the server. At present there are two implementations: -* `RestTemplateXhrTransport` uses Spring's `RestTemplate` for HTTP requests. +* `RestClientXhrTransport` uses Spring's `RestClient` for HTTP requests. * `JettyXhrTransport` uses Jetty's `HttpClient` for HTTP requests. The following example shows how to create a SockJS client and connect to a SockJS endpoint: @@ -266,7 +266,7 @@ The following example shows how to create a SockJS client and connect to a SockJ ---- List transports = new ArrayList<>(2); transports.add(new WebSocketTransport(new StandardWebSocketClient())); - transports.add(new RestTemplateXhrTransport()); + transports.add(new RestClientXhrTransport()); SockJsClient sockJsClient = new SockJsClient(transports); sockJsClient.doHandshake(new MyWebSocketHandler(), "ws://example.com:8080/sockjs"); diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransport.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransport.java new file mode 100644 index 00000000000..807d3b2ec04 --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransport.java @@ -0,0 +1,233 @@ +/* + * Copyright 2002-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.web.socket.sockjs.client; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.jspecify.annotations.Nullable; + +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.core.task.TaskExecutor; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpRequest; +import org.springframework.http.HttpStatus; +import org.springframework.http.HttpStatusCode; +import org.springframework.http.ResponseEntity; +import org.springframework.util.Assert; +import org.springframework.util.StreamUtils; +import org.springframework.web.client.HttpServerErrorException; +import org.springframework.web.client.RestClient; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.sockjs.frame.SockJsFrame; + +/** + * An {@code XhrTransport} implementation that uses a + * {@link RestClient}. + * + * @author Brian Clozel + * @author Rossen Stoyanchev + * @since 7.0.7 + */ +public class RestClientXhrTransport extends AbstractXhrTransport { + + private final RestClient restClient; + + private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); + + + public RestClientXhrTransport() { + this(RestClient.create()); + } + + public RestClientXhrTransport(RestClient restClient) { + Assert.notNull(restClient, "'restClient' is required"); + this.restClient = restClient; + } + + /** + * Return the configured {@code RestClient}. + */ + public RestClient getRestClient() { + return this.restClient; + } + + /** + * Configure the {@code TaskExecutor} to use to execute XHR receive requests. + *

By default {@link SimpleAsyncTaskExecutor + * SimpleAsyncTaskExecutor} is configured which creates a new thread every + * time the transports connects. + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + Assert.notNull(taskExecutor, "TaskExecutor must not be null"); + this.taskExecutor = taskExecutor; + } + + /** + * Return the configured {@code TaskExecutor}. + */ + public TaskExecutor getTaskExecutor() { + return this.taskExecutor; + } + + + @Override + protected void connectInternal(final TransportRequest transportRequest, final WebSocketHandler handler, + final URI receiveUrl, final HttpHeaders handshakeHeaders, final XhrClientSockJsSession session, + final CompletableFuture connectFuture) { + + getTaskExecutor().execute(() -> { + final AtomicBoolean handshakePerformed = new AtomicBoolean(); + XhrResponseReader responseReader = new XhrResponseReader(session); + while (true) { + if (session.isDisconnected()) { + session.afterTransportClosed(null); + break; + } + try { + if (logger.isTraceEnabled()) { + logger.trace("Starting XHR receive request, url=" + receiveUrl); + } + getRestClient().post().uri(receiveUrl) + .headers(headers -> { + if (handshakePerformed.compareAndSet(false, true)) { + headers.putAll(handshakeHeaders); + } + else { + headers.putAll(transportRequest.getHttpRequestHeaders()); + } + }) + .exchange(responseReader, false); + } + catch (Exception ex) { + if (!connectFuture.isDone()) { + connectFuture.completeExceptionally(ex); + } + else { + session.handleTransportError(ex); + session.afterTransportClosed(new CloseStatus(1006, ex.getMessage())); + } + break; + } + } + }); + } + + @Override + protected ResponseEntity executeInfoRequestInternal(URI infoUrl, HttpHeaders headers) { + return nonNull(this.restClient.get() + .uri(infoUrl) + .headers(httpHeaders -> httpHeaders.addAll(headers)) + .exchange(textResponseExchangeFunction)); + } + + @Override + public ResponseEntity executeSendRequestInternal(URI url, HttpHeaders headers, TextMessage message) { + return nonNull(this.restClient.post() + .uri(url) + .headers(httpHeaders -> httpHeaders.addAll(headers)) + .body(message.getPayload()) + .exchange(textResponseExchangeFunction)); + } + + private static T nonNull(@Nullable T result) { + Assert.state(result != null, "No result"); + return result; + } + + /** + * A simple ExchangeFunction that reads the body into a String. + */ + private static final RestClient.RequestHeadersSpec.ExchangeFunction> textResponseExchangeFunction = + (clientRequest, clientResponse) -> { + String body = StreamUtils.copyToString(clientResponse.getBody(), SockJsFrame.CHARSET); + return ResponseEntity.status(clientResponse.getStatusCode()).headers(clientResponse.getHeaders()).body(body); + }; + + + /** + * Splits the body of an HTTP response into SockJS frames and delegates those + * to an {@link XhrClientSockJsSession}. + */ + private class XhrResponseReader implements RestClient.RequestHeadersSpec.ExchangeFunction { + + private final XhrClientSockJsSession sockJsSession; + + public XhrResponseReader(XhrClientSockJsSession sockJsSession) { + this.sockJsSession = sockJsSession; + } + + @Override + public Void exchange(HttpRequest clientRequest, RestClient.RequestHeadersSpec.ConvertibleClientHttpResponse clientResponse) throws IOException { + HttpStatusCode httpStatus = clientResponse.getStatusCode(); + if (httpStatus != HttpStatus.OK) { + throw new HttpServerErrorException( + httpStatus, "response.getStatusCode().getStatusText()", clientResponse.getHeaders(), null, null); + } + if (logger.isTraceEnabled()) { + logger.trace("XHR receive headers: " + clientResponse.getHeaders()); + } + InputStream is = clientResponse.getBody(); + ByteArrayOutputStream os = new ByteArrayOutputStream(); + while (true) { + if (this.sockJsSession.isDisconnected()) { + if (logger.isDebugEnabled()) { + logger.debug("SockJS sockJsSession closed, closing response."); + } + clientResponse.close(); + break; + } + int b = is.read(); + if (b == -1) { + if (os.size() > 0) { + handleFrame(os); + } + if (logger.isTraceEnabled()) { + logger.trace("XHR receive completed"); + } + break; + } + if (b == '\n') { + handleFrame(os); + } + else { + os.write(b); + } + } + return null; + } + + private void handleFrame(ByteArrayOutputStream os) { + String content = os.toString(SockJsFrame.CHARSET); + os.reset(); + if (logger.isTraceEnabled()) { + logger.trace("XHR receive content: " + content); + } + if (!PRELUDE.equals(content)) { + this.sockJsSession.handleFrame(content); + } + } + } + +} diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/SockJsClient.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/SockJsClient.java index 035b77b52c0..c89ec42ea21 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/SockJsClient.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/client/SockJsClient.java @@ -94,7 +94,7 @@ public class SockJsClient implements WebSocketClient, Lifecycle { *

If the list includes an {@link XhrTransport} (or more specifically an * implementation of {@link InfoReceiver}) the instance is used to initialize * the {@link #setInfoReceiver(InfoReceiver) infoReceiver} property, or - * otherwise is defaulted to {@link RestTemplateXhrTransport}. + * otherwise is defaulted to {@link RestClientXhrTransport}. * @param transports the (non-empty) list of transports to use */ @SuppressWarnings("removal") @@ -116,7 +116,7 @@ public class SockJsClient implements WebSocketClient, Lifecycle { return infoReceiver; } } - return new RestTemplateXhrTransport(); + return new RestClientXhrTransport(); } @@ -148,7 +148,7 @@ public class SockJsClient implements WebSocketClient, Lifecycle { *

If the list of transports provided to the constructor contained an * {@link XhrTransport} or an implementation of {@link InfoReceiver} that * instance would have been used to initialize this property, or otherwise - * it defaults to {@link RestTemplateXhrTransport}. + * it defaults to {@link RestClientXhrTransport}. * @param infoReceiver the transport to use for the SockJS "Info" request */ public void setInfoReceiver(InfoReceiver infoReceiver) { diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransportTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransportTests.java new file mode 100644 index 00000000000..0075058d42b --- /dev/null +++ b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/RestClientXhrTransportTests.java @@ -0,0 +1,235 @@ +/* + * Copyright 2002-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.web.socket.sockjs.client; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.util.Arrays; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingDeque; + +import org.junit.jupiter.api.Test; + +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.client.ClientHttpRequest; +import org.springframework.http.client.ClientHttpRequestFactory; +import org.springframework.http.client.ClientHttpResponse; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.simp.stomp.StompCommand; +import org.springframework.messaging.simp.stomp.StompEncoder; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.web.client.HttpServerErrorException; +import org.springframework.web.client.RestClient; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.sockjs.frame.JacksonJsonSockJsMessageCodec; +import org.springframework.web.socket.sockjs.frame.SockJsFrame; +import org.springframework.web.socket.sockjs.transport.TransportType; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +/** + * Tests for {@link RestClientXhrTransport}. + * + * @author Brian Clozel + * @author Rossen Stoyanchev + */ +class RestClientXhrTransportTests { + + private static final JacksonJsonSockJsMessageCodec CODEC = new JacksonJsonSockJsMessageCodec(); + + private final WebSocketHandler webSocketHandler = mock(); + + + @Test + void connectReceiveAndClose() throws Exception { + String body = """ + o + a["foo"] + c[3000,"Go away!"]"""; + ClientHttpResponse response = response(HttpStatus.OK, body); + connect(response).get(); + + verify(this.webSocketHandler).afterConnectionEstablished(any()); + verify(this.webSocketHandler).handleMessage(any(), eq(new TextMessage("foo"))); + verify(this.webSocketHandler).afterConnectionClosed(any(), eq(new CloseStatus(3000, "Go away!"))); + verifyNoMoreInteractions(this.webSocketHandler); + } + + @Test + void connectReceiveAndCloseWithPrelude() throws Exception { + String prelude = "h".repeat(2048); + String body = """ + %s + o + a["foo"] + c[3000,"Go away!"]""".formatted(prelude); + ClientHttpResponse response = response(HttpStatus.OK, body); + connect(response); + + verify(this.webSocketHandler).afterConnectionEstablished(any()); + verify(this.webSocketHandler).handleMessage(any(), eq(new TextMessage("foo"))); + verify(this.webSocketHandler).afterConnectionClosed(any(), eq(new CloseStatus(3000, "Go away!"))); + verifyNoMoreInteractions(this.webSocketHandler); + } + + @Test + void connectReceiveAndCloseWithStompFrame() throws Exception { + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.SEND); + accessor.setDestination("/destination"); + MessageHeaders headers = accessor.getMessageHeaders(); + Message message = MessageBuilder.createMessage("body".getBytes(UTF_8), headers); + byte[] bytes = new StompEncoder().encode(message); + TextMessage textMessage = new TextMessage(bytes); + SockJsFrame frame = SockJsFrame.messageFrame(new JacksonJsonSockJsMessageCodec(), textMessage.getPayload()); + + String body = """ + o + %s + c[3000,"Go away!"]""".formatted(frame.getContent()); + ClientHttpResponse response = response(HttpStatus.OK, body); + connect(response); + + verify(this.webSocketHandler).afterConnectionEstablished(any()); + verify(this.webSocketHandler).handleMessage(any(), eq(textMessage)); + verify(this.webSocketHandler).afterConnectionClosed(any(), eq(new CloseStatus(3000, "Go away!"))); + verifyNoMoreInteractions(this.webSocketHandler); + } + + @Test + void connectFailure() { + final HttpServerErrorException expected = new HttpServerErrorException(HttpStatus.INTERNAL_SERVER_ERROR); + FailingClientHttpRequestFactory requestFactory = new FailingClientHttpRequestFactory(expected); + final CountDownLatch latch = new CountDownLatch(1); + connect(requestFactory).whenComplete((result, ex) -> { + if (ex == expected) { + latch.countDown(); + } + }); + verifyNoMoreInteractions(this.webSocketHandler); + } + + @Test + void errorResponseStatus() throws Exception { + connect(response(HttpStatus.OK, "o\n"), response(HttpStatus.INTERNAL_SERVER_ERROR, "Oops")); + + verify(this.webSocketHandler).afterConnectionEstablished(any()); + verify(this.webSocketHandler).handleTransportError(any(), any()); + verify(this.webSocketHandler).afterConnectionClosed(any(), any()); + verifyNoMoreInteractions(this.webSocketHandler); + } + + @Test + void responseClosedAfterDisconnected() throws Exception { + String body = """ + o + c[3000,"Go away!"] + a["foo"] + """; + ClientHttpResponse response = response(HttpStatus.OK, body); + connect(response); + + verify(this.webSocketHandler).afterConnectionEstablished(any()); + verify(this.webSocketHandler).afterConnectionClosed(any(), any()); + verifyNoMoreInteractions(this.webSocketHandler); + verify(response).close(); + } + + private CompletableFuture connect(ClientHttpResponse... responses) { + return connect(new TestClientHttpRequestFactory(responses)); + } + + private CompletableFuture connect(ClientHttpRequestFactory requestFactory) { + RestClient restClient = RestClient.builder().requestFactory(requestFactory).build(); + RestClientXhrTransport transport = new RestClientXhrTransport(restClient); + transport.setTaskExecutor(new SyncTaskExecutor()); + + SockJsUrlInfo urlInfo = new SockJsUrlInfo(URI.create("https://example.com")); + HttpHeaders headers = new HttpHeaders(); + headers.add("h-foo", "h-bar"); + TransportRequest request = new DefaultTransportRequest(urlInfo, headers, headers, + transport, TransportType.XHR, CODEC); + + return transport.connectAsync(request, this.webSocketHandler); + } + + private ClientHttpResponse response(HttpStatus status, String body) throws IOException { + ClientHttpResponse response = mock(); + InputStream inputStream = getInputStream(body); + given(response.getStatusCode()).willReturn(status); + given(response.getBody()).willReturn(inputStream); + return response; + } + + private InputStream getInputStream(String content) { + byte[] bytes = content.getBytes(UTF_8); + return new ByteArrayInputStream(bytes); + } + + private static class TestClientHttpRequestFactory implements ClientHttpRequestFactory { + + private Queue responses = new LinkedBlockingDeque<>(); + + private TestClientHttpRequestFactory(ClientHttpResponse... responses) { + this.responses.addAll(Arrays.asList(responses)); + } + + @Override + public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException { + ClientHttpRequest request = mock(); + given(request.getHeaders()).willReturn(new HttpHeaders()); + given(request.execute()).willReturn(this.responses.remove()); + return request; + } + + } + + private static class FailingClientHttpRequestFactory implements ClientHttpRequestFactory { + + private final Exception expected; + + private FailingClientHttpRequestFactory(Exception expected) { + this.expected = expected; + } + + @Override + public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException { + ClientHttpRequest request = mock(); + given(request.getHeaders()).willReturn(new HttpHeaders()); + given(request.execute()).willThrow(this.expected); + return request; + } + } + +}