From e08ea45ebe1dc01b6fd9f3a541564655ccded019 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 16 Jun 2020 15:37:38 -0400 Subject: [PATCH] Websocket supplier/consumer - source/sink --- common/function-test-support/pom.xml | 4 + .../WebsocketConsumerClientHandler.java | 60 +++++ consumer/websocket-consumer/README.adoc | 25 +++ consumer/websocket-consumer/pom.xml | 60 +++++ .../WebsocketConsumerConfiguration.java | 111 +++++++++ .../WebsocketConsumerProperties.java | 114 ++++++++++ .../websocket/WebsocketConsumerServer.java | 112 +++++++++ .../WebsocketConsumerServerHandler.java | 212 ++++++++++++++++++ .../WebsocketConsumerServerInitializer.java | 90 ++++++++ .../WebsocketConsumerTraceEndpoint.java | 68 ++++++ .../trace/InMemoryTraceRepository.java | 85 +++++++ .../fn/consumer/websocket/trace/Trace.java | 56 +++++ .../websocket/WebsocketConsumerTests.java | 162 +++++++++++++ pom.xml | 2 + supplier/websocket-supplier/README.adoc | 32 +++ supplier/websocket-supplier/pom.xml | 56 +++++ .../WebsocketSupplierConfiguration.java | 86 +++++++ .../WebsocketSupplierProperties.java | 96 ++++++++ .../websocket/WebsocketSupplierTests.java | 102 +++++++++ 19 files changed, 1533 insertions(+) create mode 100644 common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/websocket/WebsocketConsumerClientHandler.java create mode 100644 consumer/websocket-consumer/README.adoc create mode 100644 consumer/websocket-consumer/pom.xml create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerConfiguration.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServer.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerHandler.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerInitializer.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/actuator/WebsocketConsumerTraceEndpoint.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/InMemoryTraceRepository.java create mode 100644 consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/Trace.java create mode 100644 consumer/websocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerTests.java create mode 100644 supplier/websocket-supplier/README.adoc create mode 100644 supplier/websocket-supplier/pom.xml create mode 100644 supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java create mode 100644 supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierProperties.java create mode 100644 supplier/websocket-supplier/src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java diff --git a/common/function-test-support/pom.xml b/common/function-test-support/pom.xml index 9fd267dc..d259b9a2 100644 --- a/common/function-test-support/pom.xml +++ b/common/function-test-support/pom.xml @@ -57,6 +57,10 @@ spring-data-geode true + + org.springframework + spring-websocket + diff --git a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/websocket/WebsocketConsumerClientHandler.java b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/websocket/WebsocketConsumerClientHandler.java new file mode 100644 index 00000000..8a857d0f --- /dev/null +++ b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/websocket/WebsocketConsumerClientHandler.java @@ -0,0 +1,60 @@ +/* + * Copyright 2014-2020 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.cloud.fn.test.support.websocket; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.AbstractWebSocketHandler; + +public class WebsocketConsumerClientHandler extends AbstractWebSocketHandler { + + final List receivedMessages = new ArrayList<>(); + + final int waitMessageCount; + + final CountDownLatch latch; + + final long timeout; + + final String id; + + public WebsocketConsumerClientHandler(String id, int waitMessageCount, long timeout) { + this.id = id; + this.waitMessageCount = waitMessageCount; + this.latch = new CountDownLatch(waitMessageCount); + this.timeout = timeout; + } + + @Override + public void handleTextMessage(WebSocketSession session, TextMessage message) { + receivedMessages.add(message.getPayload()); + latch.countDown(); + } + + public void await() throws InterruptedException { + latch.await(timeout, TimeUnit.MILLISECONDS); + } + + public List getReceivedMessages() { + return receivedMessages; + } +} diff --git a/consumer/websocket-consumer/README.adoc b/consumer/websocket-consumer/README.adoc new file mode 100644 index 00000000..16669f23 --- /dev/null +++ b/consumer/websocket-consumer/README.adoc @@ -0,0 +1,25 @@ +# Websocket Consumer + +A consumer that allows you to send messages using websocket. + +## Beans for injection + +You can import `WebsocketConsumerConfiguration` in the application and then inject the following bean. + +`Consumer> websocketConsumer` + +You can use `websocketConsumer` as a qualifier when injecting. + +## Configuration Options + +All configuration properties are prefixed with `websocket.consumer`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java[WebsocketConsumerProperties]. + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/consumer/websocket[test suite] for the various ways, this consumer is used. + +## Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/websocket-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a TCP sink. \ No newline at end of file diff --git a/consumer/websocket-consumer/pom.xml b/consumer/websocket-consumer/pom.xml new file mode 100644 index 00000000..5f28f012 --- /dev/null +++ b/consumer/websocket-consumer/pom.xml @@ -0,0 +1,60 @@ + + + 4.0.0 + websocket-consumer + 1.0.0-SNAPSHOT + websocket-consumer + websocket consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework + spring-messaging + + + org.springframework + spring-websocket + + + io.netty + netty-all + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.boot + spring-boot-starter-web + test + + + org.springframework.cloud.fn + function-test-support + ${spring-cloud-fn.version} + test + + + + diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerConfiguration.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerConfiguration.java new file mode 100644 index 00000000..e7597ee4 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerConfiguration.java @@ -0,0 +1,111 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.function.Consumer; + +import javax.annotation.PostConstruct; + +import io.netty.channel.Channel; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.consumer.websocket.actuator.WebsocketConsumerTraceEndpoint; +import org.springframework.cloud.fn.consumer.websocket.trace.InMemoryTraceRepository; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +import org.springframework.messaging.simp.SimpMessageHeaderAccessor; +import org.springframework.messaging.simp.SimpMessageType; + +/** + * @author Oliver Moser + * @author Gary Russell + * @author Artem Bilan + */ +@Configuration +@EnableConfigurationProperties(WebsocketConsumerProperties.class) +public class WebsocketConsumerConfiguration { + + + private static final Log logger = LogFactory.getLog(WebsocketConsumerConfiguration.class); + + private final InMemoryTraceRepository websocketTraceRepository = new InMemoryTraceRepository(); + + @Value("${endpoints.websocketconsumertrace.enabled:false}") + private boolean traceEndpointEnabled; + + @PostConstruct + public void init() throws InterruptedException { + server().run(); + } + + @Bean + public WebsocketConsumerServer server() { + return new WebsocketConsumerServer(); + } + + @Bean + public WebsocketConsumerServerInitializer initializer() { + return new WebsocketConsumerServerInitializer(this.websocketTraceRepository); + } + + @Bean + @ConditionalOnProperty(value = "endpoints.websocketsinktrace.enabled", havingValue = "true") + public WebsocketConsumerTraceEndpoint websocketTraceEndpoint() { + return new WebsocketConsumerTraceEndpoint(this.websocketTraceRepository); + } + + @Bean + public Consumer> websocketConsumer() { + return message -> { + if (logger.isTraceEnabled()) { + logger.trace("Handling message: " + message); + } + SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message); + headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE); + String messagePayload = message.getPayload().toString(); + for (Channel channel : WebsocketConsumerServer.channels) { + if (logger.isTraceEnabled()) { + logger.trace(String.format("Writing message %s to channel %s", messagePayload, channel.localAddress())); + } + + channel.write(new TextWebSocketFrame(messagePayload)); + channel.flush(); + } + + if (this.traceEndpointEnabled) { + addMessageToTraceRepository(message); + } + }; + } + + private void addMessageToTraceRepository(Message message) { + Map trace = new LinkedHashMap<>(); + trace.put("type", "text"); + trace.put("direction", "out"); + trace.put("id", message.getHeaders().getId()); + trace.put("payload", message.getPayload().toString()); + this.websocketTraceRepository.add(trace); + } +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java new file mode 100644 index 00000000..4c881c36 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerProperties.java @@ -0,0 +1,114 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import io.netty.handler.logging.LogLevel; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * @author Oliver Moser + * @author Gary Russell + */ +@ConfigurationProperties("websocket.consumer") +public class WebsocketConsumerProperties { + + /** + * Default log level. + */ + public static final String DEFAULT_LOGLEVEL = LogLevel.WARN.toString(); + + /** + * Default path. + */ + public static final String DEFAULT_PATH = "/websocket"; + + /** + * Default number of threads. + */ + public static final int DEFAULT_THREADS = 1; + + /** + * Default port. + */ + public static final int DEFAULT_PORT = 9292; + + /** + * whether or not to create a {@link io.netty.handler.ssl.SslContext}. + */ + boolean ssl; + + /** + * the port on which the Netty server listens. Default is 9292 + */ + int port = DEFAULT_PORT; + + /** + * the number of threads for the Netty {@link io.netty.channel.EventLoopGroup}. Default is 1 + */ + int threads = DEFAULT_THREADS; + + /** + * the logLevel for netty channels. Default is WARN + */ + String logLevel = DEFAULT_LOGLEVEL; + + /** + * the path on which a WebsocketSink consumer needs to connect. Default is /websocket + */ + String path = DEFAULT_PATH; + + public boolean isSsl() { + return ssl; + } + + public void setSsl(boolean ssl) { + this.ssl = ssl; + } + + public int getPort() { + return port; + } + + public void setPort(int port) { + this.port = port; + } + + public int getThreads() { + return threads; + } + + public void setThreads(int threads) { + this.threads = threads; + } + + public String getLogLevel() { + return logLevel; + } + + public void setLogLevel(String logLevel) { + this.logLevel = logLevel; + } + + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServer.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServer.java new file mode 100644 index 00000000..4986ed74 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServer.java @@ -0,0 +1,112 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; + +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.logging.LogLevel; +import io.netty.handler.logging.LoggingHandler; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.annotation.Autowired; + +/** + * Bootstraps a Netty server using the {@link WebsocketConsumerServerInitializer}. Also adds + * a {@link LoggingHandler} and uses the logLevel + * from {@link WebsocketConsumerProperties#logLevel}. + * + * @author Oliver Moser + * @author Gary Russell + */ +public class WebsocketConsumerServer { + + private static final Log logger = LogFactory.getLog(WebsocketConsumerServer.class); + + static final List channels = Collections.synchronizedList(new ArrayList()); + + @Autowired + WebsocketConsumerProperties properties; + + @Autowired + WebsocketConsumerServerInitializer initializer; + + private EventLoopGroup bossGroup; + + private EventLoopGroup workerGroup; + + private int port; + + public int getPort() { + return this.port; + } + + @PostConstruct + public void init() { + bossGroup = new NioEventLoopGroup(properties.getThreads()); + workerGroup = new NioEventLoopGroup(); + } + + @PreDestroy + public void shutdown() { + bossGroup.shutdownGracefully(); + workerGroup.shutdownGracefully(); + } + + public void run() throws InterruptedException { + NioServerSocketChannel channel = (NioServerSocketChannel) new ServerBootstrap().group(bossGroup, workerGroup) + .channel(NioServerSocketChannel.class) + .handler(new LoggingHandler(nettyLogLevel())) + .childHandler(initializer) + .bind(properties.getPort()) + .sync() + .channel(); + this.port = channel.localAddress().getPort(); + dumpProperties(); + } + + private void dumpProperties() { + logger.info("███████████████████████████████████████████████████████████"); + logger.info(" >> websocket-sink config << "); + logger.info(""); + logger.info(String.format("port: %s", this.port)); + logger.info(String.format("ssl: %s", this.properties.isSsl())); + logger.info(String.format("path: %s", this.properties.getPath())); + logger.info(String.format("logLevel: %s", this.properties.getLogLevel())); + logger.info(String.format("threads: %s", this.properties.getThreads())); + logger.info(""); + logger.info("████████████████████████████████████████████████████████████"); + } + + // + // HELPERS + // + private LogLevel nettyLogLevel() { + return LogLevel.valueOf(properties.getLogLevel().toUpperCase()); + } + +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerHandler.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerHandler.java new file mode 100644 index 00000000..a1856df6 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerHandler.java @@ -0,0 +1,212 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import java.util.LinkedHashMap; +import java.util.Map; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.FullHttpResponse; +import io.netty.handler.codec.http.HttpUtil; +import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; +import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketFrame; +import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker; +import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory; +import io.netty.util.CharsetUtil; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.cloud.fn.consumer.websocket.trace.InMemoryTraceRepository; + +import static io.netty.handler.codec.http.HttpHeaderNames.HOST; +import static io.netty.handler.codec.http.HttpMethod.GET; +import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST; +import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN; +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; + +/** + * Handles handshakes and messages. Based on the Netty websocket examples. + * + * @author Netty Project + * @author Oliver Moser + * @author Gary Russell + * @author Artem Bilan + */ +public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler { + + private static final Log logger = LogFactory.getLog(WebsocketConsumerServerHandler.class); + + private final boolean traceEnabled; + + private final InMemoryTraceRepository websocketTraceRepository; + + private final WebsocketConsumerProperties properties; + + private WebSocketServerHandshaker handshaker; + + public WebsocketConsumerServerHandler(InMemoryTraceRepository websocketTraceRepository, + WebsocketConsumerProperties properties, + boolean traceEnabled) { + + this.websocketTraceRepository = websocketTraceRepository; + this.properties = properties; + this.traceEnabled = traceEnabled; + } + + @Override + public void channelRead0(ChannelHandlerContext ctx, Object msg) { + if (msg instanceof FullHttpRequest) { + handleHttpRequest(ctx, (FullHttpRequest) msg); + } + else if (msg instanceof WebSocketFrame) { + handleWebSocketFrame(ctx, (WebSocketFrame) msg); + } + } + + @Override + public void channelReadComplete(ChannelHandlerContext ctx) { + ctx.flush(); + } + + private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { + // Handle a bad request. + if (!req.decoderResult().isSuccess()) { + logger.warn(String.format("Bad request: %s", req.uri())); + sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST)); + return; + } + + // Allow only GET methods. + if (req.method() != GET) { + logger.warn(String.format("Unsupported HTTP method: %s", req.method())); + sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN)); + return; + } + + // enable subclasses to do additional processing + if (!additionalHttpRequestHandler(ctx, req)) { + return; + } + + // Handshake + WebSocketServerHandshakerFactory wsFactory + = new WebSocketServerHandshakerFactory(getWebSocketLocation(req), null, true); + + this.handshaker = wsFactory.newHandshaker(req); + if (this.handshaker == null) { + WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel()); + } + else { + this.handshaker.handshake(ctx.channel(), req); + WebsocketConsumerServer.channels.add(ctx.channel()); + } + } + + private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { + // Check for closing frame + if (frame instanceof CloseWebSocketFrame) { + addTraceForFrame(frame, "close"); + this.handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain()); + return; + } + if (frame instanceof PingWebSocketFrame) { + addTraceForFrame(frame, "ping"); + ctx.channel().write(new PongWebSocketFrame(frame.content().retain())); + return; + } + if (!(frame instanceof TextWebSocketFrame)) { + throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass() + .getName())); + } + + // todo [om] think about BinaryWebsocketFrame + + handleTextWebSocketFrameInternal((TextWebSocketFrame) frame, ctx); + } + + private boolean additionalHttpRequestHandler(ChannelHandlerContext ctx, FullHttpRequest req) { + // implement other HTTP request logic + return true; // continue processing + } + + // simple echo implementation + private void handleTextWebSocketFrameInternal(TextWebSocketFrame frame, ChannelHandlerContext ctx) { + if (logger.isTraceEnabled()) { + logger.trace(String.format("%s received %s", ctx.channel(), frame.text())); + } + + addTraceForFrame(frame, "text"); + ctx.channel().write(new TextWebSocketFrame("Echo: " + frame.text())); + } + + // add trace information for received frame + private void addTraceForFrame(WebSocketFrame frame, String type) { + Map trace = new LinkedHashMap<>(); + trace.put("type", type); + trace.put("direction", "in"); + if (frame instanceof TextWebSocketFrame) { + trace.put("payload", ((TextWebSocketFrame) frame).text()); + } + + if (this.traceEnabled) { + this.websocketTraceRepository.add(trace); + } + } + + private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) { + // Generate an error page if response getStatus code is not OK (200). + if (res.status().code() != 200) { + ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8); + res.content().writeBytes(buf); + buf.release(); + HttpUtil.setContentLength(res, res.content().readableBytes()); + } + + // Send the response and close the connection if necessary. + ChannelFuture f = ctx.channel().writeAndFlush(res); + if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) { + f.addListener(ChannelFutureListener.CLOSE); + } + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + logger.error("Websocket error", cause); + cause.printStackTrace(); + ctx.close(); + } + + private String getWebSocketLocation(FullHttpRequest req) { + String location = req.headers().get(HOST) + this.properties.getPath(); + if (this.properties.isSsl()) { + return "wss://" + location; + } + else { + return "ws://" + location; + } + } +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerInitializer.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerInitializer.java new file mode 100644 index 00000000..ffdefc02 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerServerInitializer.java @@ -0,0 +1,90 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import java.security.cert.CertificateException; + +import javax.net.ssl.SSLException; + +import io.netty.channel.ChannelInitializer; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.ssl.SslContext; +import io.netty.handler.ssl.SslContextBuilder; +import io.netty.handler.ssl.util.SelfSignedCertificate; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.cloud.fn.consumer.websocket.trace.InMemoryTraceRepository; + +/** + * Does some basic initialization and setup. + * + *
    + *
  • Configure the {@link SslContext} based on {@link WebsocketConsumerProperties#ssl}
    • + *
  • add the {@link WebsocketConsumerServerHandler} to the underlying {@link ChannelPipeline}
    • + *
+ * + * @author Oliver Moser + * @author Gary Russell + * @author Artem Bilan + */ +public class WebsocketConsumerServerInitializer extends ChannelInitializer { + + /** + * Max content length. + */ + public static final int MAX_CONTENT_LENGTH = 65536; + + private final InMemoryTraceRepository traceRepository; + + @Autowired + private WebsocketConsumerProperties properties; + + @Value("${endpoints.websocketsinktrace.enabled:false}") + private boolean traceEnabled; + + public WebsocketConsumerServerInitializer(InMemoryTraceRepository traceRepository) { + this.traceRepository = traceRepository; + } + + @Override + public void initChannel(SocketChannel ch) throws Exception { + ChannelPipeline pipeline = ch.pipeline(); + + final SslContext sslCtx = configureSslContext(); + if (sslCtx != null) { + pipeline.addLast(sslCtx.newHandler(ch.alloc())); + } + + pipeline.addLast(new HttpServerCodec()); + pipeline.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH)); + pipeline.addLast(new WebsocketConsumerServerHandler(this.traceRepository, this.properties, this.traceEnabled)); + } + + private SslContext configureSslContext() throws CertificateException, SSLException { + if (this.properties.isSsl()) { + SelfSignedCertificate ssc = new SelfSignedCertificate(); + return SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build(); + } + else { + return null; + } + } +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/actuator/WebsocketConsumerTraceEndpoint.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/actuator/WebsocketConsumerTraceEndpoint.java new file mode 100644 index 00000000..3f4f3fbd --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/actuator/WebsocketConsumerTraceEndpoint.java @@ -0,0 +1,68 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket.actuator; + +import java.util.List; + +import javax.annotation.PostConstruct; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.boot.actuate.endpoint.annotation.Endpoint; +import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.fn.consumer.websocket.trace.InMemoryTraceRepository; +import org.springframework.cloud.fn.consumer.websocket.trace.Trace; + +/** + * Simple Spring Boot Actuator {@link Endpoint} implementation that + * provides access to Websocket messages last sent/received. + * + * @author Oliver Moser + * @author Artem Bilan + */ +@ConfigurationProperties(prefix = "endpoints.websocketconsumertrace") +@Endpoint(id = "websocketconsumertrace") +public class WebsocketConsumerTraceEndpoint { + + private static final Log logger = LogFactory.getLog(WebsocketConsumerTraceEndpoint.class); + + private boolean enabled; + + private final InMemoryTraceRepository repository; + + public WebsocketConsumerTraceEndpoint(InMemoryTraceRepository repository) { + this.repository = repository; + logger.info(String.format("/websocketsinktrace enabled: %b", this.enabled)); + } + + @PostConstruct + public void init() { + + } + + @ReadOperation + public List traces() { + return this.repository.findAll(); + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/InMemoryTraceRepository.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/InMemoryTraceRepository.java new file mode 100644 index 00000000..bdd47f11 --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/InMemoryTraceRepository.java @@ -0,0 +1,85 @@ +/* + * Copyright 2018-2020 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.cloud.fn.consumer.websocket.trace; + +import java.util.Collections; +import java.util.Date; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +/** + * A repository for {@link Trace}s. + * + * It is a copy of {@code InMemoryTraceRepository} from Spring Boot 1.5.x. + * Since Spring Boot 2.0 traces are only available for HTTP. + * + * @author Dave Syer + * @author Olivier Bourgain + * @author Artem Bilan + * + * @since 2.0 + */ +public class InMemoryTraceRepository { + + private int capacity = 100; + + private boolean reverse = true; + + private final List traces = new LinkedList<>(); + + /** + * Flag to say that the repository lists traces in reverse order. + * @param reverse flag value (default true) + */ + public void setReverse(boolean reverse) { + synchronized (this.traces) { + this.reverse = reverse; + } + } + + /** + * Set the capacity of the in-memory repository. + * @param capacity the capacity + */ + public void setCapacity(int capacity) { + synchronized (this.traces) { + this.capacity = capacity; + } + } + + public List findAll() { + synchronized (this.traces) { + return Collections.unmodifiableList(this.traces); + } + } + + public void add(Map map) { + Trace trace = new Trace(new Date(), map); + synchronized (this.traces) { + while (this.traces.size() >= this.capacity) { + this.traces.remove(this.reverse ? this.capacity - 1 : 0); + } + if (this.reverse) { + this.traces.add(0, trace); + } + else { + this.traces.add(trace); + } + } + } +} diff --git a/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/Trace.java b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/Trace.java new file mode 100644 index 00000000..636fbe3f --- /dev/null +++ b/consumer/websocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/websocket/trace/Trace.java @@ -0,0 +1,56 @@ +/* + * Copyright 2018-2020 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.cloud.fn.consumer.websocket.trace; + +import java.util.Date; +import java.util.Map; + +import org.springframework.util.Assert; + +/** + * A value object representing a trace event: at a particular time with a simple (map) + * information. Can be used for analyzing contextual information such as HTTP headers. + * + *

It is a copy of {@code InMemoryTraceRepository} from Spring Boot 1.5.x. + * Since Spring Boot 2.0 traces are only available for HTTP. + * + * @author Dave Syer + * @author Artem Bilan + * + * @since 2.0 + */ +public class Trace { + + private final Date timestamp; + + private final Map info; + + public Trace(Date timestamp, Map info) { + Assert.notNull(timestamp, "Timestamp must not be null"); + Assert.notNull(info, "Info must not be null"); + this.timestamp = timestamp; + this.info = info; + } + + public Date getTimestamp() { + return this.timestamp; + } + + public Map getInfo() { + return this.info; + } +} diff --git a/consumer/websocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerTests.java b/consumer/websocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerTests.java new file mode 100644 index 00000000..9bbb6e02 --- /dev/null +++ b/consumer/websocket-consumer/src/test/java/org/springframework/cloud/fn/consumer/websocket/WebsocketConsumerTests.java @@ -0,0 +1,162 @@ +/* + * Copyright 2014-2020 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.cloud.fn.consumer.websocket; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.function.Consumer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.fn.test.support.websocket.WebsocketConsumerClientHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Oliver Moser + * @author Gary Russell + * @author Artem Bilan + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = { + "websocket.consumer.port=0", + "websocket.consumer.path=/some_websocket_path", + "websocket.consumer.logLevel=DEBUG", + "websocket.consumer.threads=2" + }) +@DirtiesContext +public class WebsocketConsumerTests { + + public static final int TIMEOUT = 10000; + + public static final int MESSAGE_COUNT = 100; + + public static final int CLIENT_COUNT = 10; + + @Autowired + private WebsocketConsumerProperties properties; + + @Autowired + private WebsocketConsumerServer consumerServer; + + @Autowired + Consumer> websocketConsumer; + + @Test + public void checkCmdlineArgs() { + assertThat(properties.getPath()).isEqualTo("/some_websocket_path"); + assertThat(properties.getPort()).isEqualTo((0)); + assertThat(properties.getLogLevel()).isEqualTo(("DEBUG")); + assertThat(properties.getThreads()).isEqualTo((2)); + } + + @Test + @Timeout(TIMEOUT) + public void testMultipleMessageSingleSubscriber() throws Exception { + WebsocketConsumerClientHandler handler = new WebsocketConsumerClientHandler("handler_0", MESSAGE_COUNT, TIMEOUT); + doHandshake(handler); + + List messagesToSend = submitMultipleMessages(MESSAGE_COUNT); + handler.await(); + + assertThat(handler.getReceivedMessages().size()).isEqualTo(MESSAGE_COUNT); + messagesToSend.forEach(s -> assertThat(handler.getReceivedMessages().contains(s)).isTrue()); + } + + @Test + @Timeout(TIMEOUT) + public void testSingleMessageMultipleSubscribers() throws Exception { + + // create multiple handlers + List handlers = createHandlerList(CLIENT_COUNT, 1); + + // submit a single message + String payload = UUID.randomUUID().toString(); + websocketConsumer.accept(MessageBuilder.withPayload(payload).build()); + + // await completion on each handler + for (WebsocketConsumerClientHandler handler : handlers) { + handler.await(); + assertThat(handler.getReceivedMessages().size()).isEqualTo(1); + assertThat(handler.getReceivedMessages().get(0)).isEqualTo(payload); + } + } + + @Test + @Timeout(TIMEOUT) + public void testMultipleMessagesMultipleSubscribers() throws Exception { + + // create multiple handlers + List handlers = createHandlerList(CLIENT_COUNT, MESSAGE_COUNT); + + // submit mulitple message + List messagesReceived = submitMultipleMessages(MESSAGE_COUNT); + + // wait on each handle + for (WebsocketConsumerClientHandler handler : handlers) { + handler.await(); + assertThat(handler.getReceivedMessages().size()).isEqualTo(messagesReceived.size()); + assertThat(handler.getReceivedMessages()).isEqualTo(messagesReceived); + } + } + + private WebSocketSession doHandshake(WebsocketConsumerClientHandler handler) + throws InterruptedException, ExecutionException { + String wsEndpoint = "ws://localhost:" + this.consumerServer.getPort() + this.properties.getPath(); + return new StandardWebSocketClient().doHandshake(handler, wsEndpoint).get(); + } + + private List submitMultipleMessages(int messageCount) { + List messagesToSend = new ArrayList<>(messageCount); + for (int i = 0; i < messageCount; i++) { + String message = "message_" + i; + messagesToSend.add(message); + websocketConsumer.accept(MessageBuilder.withPayload(message).build()); + } + + return messagesToSend; + } + + private List createHandlerList(int handlerCount, int messageCount) throws + InterruptedException, + ExecutionException { + + List handlers = new ArrayList<>(handlerCount); + for (int i = 0; i < handlerCount; i++) { + WebsocketConsumerClientHandler handler = new WebsocketConsumerClientHandler("handler_" + i, messageCount, TIMEOUT); + doHandshake(handler); + handlers.add(handler); + } + return handlers; + } + + @SpringBootApplication + public static class WebsocketConsumerTestApplication { + + } +} diff --git a/pom.xml b/pom.xml index df5aacfd..cad847d0 100644 --- a/pom.xml +++ b/pom.xml @@ -64,6 +64,7 @@ consumer/redis-consumer consumer/sftp-consumer consumer/tcp-consumer + consumer/websocket-consumer function/filter-function function/header-enricher-function @@ -83,6 +84,7 @@ supplier/tcp-supplier supplier/time-supplier supplier/rabbit-supplier + supplier/websocket-supplier spring-functions-parent diff --git a/supplier/websocket-supplier/README.adoc b/supplier/websocket-supplier/README.adoc new file mode 100644 index 00000000..33f546d3 --- /dev/null +++ b/supplier/websocket-supplier/README.adoc @@ -0,0 +1,32 @@ +# Websocket Supplier + +A basic websocket supplier that produced messages through web socket. +The `Supplier` uses the `WebsocketInboundChannelAdapter` from Spring Integration. +This supplier gives you a reactive stream of messages and the supplier has a signature of `Supplier>>`. +Users have to subscribe to this `Flux` and receive the data. + +## Beans for injection + +You can import the `WebsocketSupplierConfiguration` in the application and then inject the following bean. + +`websocketSupplier` + +You need to inject this as `Supplier>>`. + +You can use `websocketSupplier` as a qualifier when injecting. + +Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`. + +## Configuration Options + +All configuration properties are prefixed with `websocket.supplier`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierProperties.java[WebsocketSupplierProperties]. + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java[test suite] for the various ways, this supplier is used. + +## Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/websocket-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a File Source. \ No newline at end of file diff --git a/supplier/websocket-supplier/pom.xml b/supplier/websocket-supplier/pom.xml new file mode 100644 index 00000000..5337adcf --- /dev/null +++ b/supplier/websocket-supplier/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + websocket-supplier + 1.0.0-SNAPSHOT + websocket-supplier + websocket supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-websocket + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.boot + spring-boot-starter-web + test + + + org.springframework.boot + spring-boot-starter-security + test + + + io.projectreactor + reactor-test + test + + + + diff --git a/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java b/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java new file mode 100644 index 00000000..3a3d917e --- /dev/null +++ b/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java @@ -0,0 +1,86 @@ +/* + * Copyright 2018-2020 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.cloud.fn.supplier.websocket; + +import java.util.function.Supplier; + +import reactor.core.publisher.Flux; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.integration.websocket.IntegrationWebSocketContainer; +import org.springframework.integration.websocket.ServerWebSocketContainer; +import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter; +import org.springframework.messaging.Message; + +/** + * A supplier that receives data over WebSocket. + * + * @author Krishnaprasad A S + * @author Artem Bilan + * + */ +@Configuration +@EnableConfigurationProperties(WebsocketSupplierProperties.class) +public class WebsocketSupplierConfiguration { + + @Autowired + WebsocketSupplierProperties properties; + + @Bean + public Supplier>> websocketSupplier(WebSocketInboundChannelAdapter webSocketInboundChannelAdapter) { + return () -> Flux.from(output()) + .doOnSubscribe(subscription -> webSocketInboundChannelAdapter.start()); + } + + @Bean + public FluxMessageChannel output() { + return new FluxMessageChannel(); + } + + @Bean + @ConditionalOnMissingBean + @ConditionalOnProperty(prefix = "websocket.sockJs", name = "enable", havingValue = "true") + public ServerWebSocketContainer.SockJsServiceOptions sockJsServiceOptions() { + // TODO Expose SockJsServiceOptions as configuration properties + return new ServerWebSocketContainer.SockJsServiceOptions(); + } + + @Bean + public IntegrationWebSocketContainer serverWebSocketContainer( + ObjectProvider sockJsServiceOptions) { + return new ServerWebSocketContainer(properties.getPath()) + .setAllowedOrigins(properties.getAllowedOrigins()) + .withSockJs(sockJsServiceOptions.getIfAvailable()); + } + + @Bean + public WebSocketInboundChannelAdapter webSocketInboundChannelAdapter( + IntegrationWebSocketContainer serverWebSocketContainer) { + WebSocketInboundChannelAdapter webSocketInboundChannelAdapter = + new WebSocketInboundChannelAdapter(serverWebSocketContainer); + webSocketInboundChannelAdapter.setOutputChannel(output()); + webSocketInboundChannelAdapter.setAutoStartup(false); + return webSocketInboundChannelAdapter; + } +} diff --git a/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierProperties.java b/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierProperties.java new file mode 100644 index 00000000..3d23343c --- /dev/null +++ b/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierProperties.java @@ -0,0 +1,96 @@ +/* + * Copyright 2018-2020 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.cloud.fn.supplier.websocket; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Configuration properties available for Websocket Supplier. + * + * @author Krishnaprasad A S + * @author Artem Bilan + * + */ +@ConfigurationProperties("websocket.supplier") +public class WebsocketSupplierProperties { + + /** + * Default path. + */ + public static final String DEFAULT_PATH = "/websocket"; + + /** + * Default allowed origins. + */ + public static final String DEFAULT_ALLOWED_ORIGINS = "*"; + + /** + * The path on which server WebSocket handler is exposed. + */ + private String path = DEFAULT_PATH; + + /** + * The allowed origins. + */ + private String allowedOrigins = DEFAULT_ALLOWED_ORIGINS; + + /** + * The SockJS options. + */ + private SockJs sockJs = new SockJs(); + + public String getPath() { + return this.path; + } + + public void setPath(String path) { + this.path = path; + } + + public String getAllowedOrigins() { + return this.allowedOrigins; + } + + public void setAllowedOrigins(String allowedOrigins) { + this.allowedOrigins = allowedOrigins; + } + + public SockJs getSockJs() { + return this.sockJs; + } + + public void setSockJs(SockJs sockJs) { + this.sockJs = sockJs; + } + + public static class SockJs { + + /** + * Enable SockJS service on the server. Default is 'false' + */ + private boolean enable; + + public boolean getEnable() { + return this.enable; + } + + public void setEnable(boolean enable) { + this.enable = enable; + } + + } +} diff --git a/supplier/websocket-supplier/src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java b/supplier/websocket-supplier/src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java new file mode 100644 index 00000000..bfe5a5fe --- /dev/null +++ b/supplier/websocket-supplier/src/test/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierTests.java @@ -0,0 +1,102 @@ +/* + * Copyright 2018-2020 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.cloud.fn.supplier.websocket; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.function.Supplier; + +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.security.SecurityProperties; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.web.server.LocalServerPort; +import org.springframework.http.HttpHeaders; +import org.springframework.integration.websocket.ClientWebSocketContainer; +import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.util.Base64Utils; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, + properties = "websocket.supplier.path=/some_websocket_path") +@DirtiesContext +public class WebsocketSupplierTests { + + @Autowired + private Supplier>> websocketSupplier; + + @LocalServerPort + private int port; + + @Autowired + private WebsocketSupplierProperties properties; + + @Autowired + private SecurityProperties securityProperties; + + private final String messageString = "foo"; + + @Test + public void checkCmdlineArgs() { + assertThat(this.properties.getPath()).isEqualTo("/some_websocket_path"); + assertThat(this.properties.getAllowedOrigins()).isEqualTo("*"); + } + + @Test + public void testBasicFlow() throws IOException { + final Flux> messageFlux = websocketSupplier.get(); + final StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext((message) -> { + assertThat(message.getPayload()) + .isEqualTo(messageString); + } + ) + .thenCancel() + .verifyLater(); + StandardWebSocketClient webSocketClient = new StandardWebSocketClient(); + ClientWebSocketContainer clientWebSocketContainer = + new ClientWebSocketContainer(webSocketClient, "ws://localhost:{port}{path}", + this.port, + this.properties.getPath()); + + HttpHeaders httpHeaders = new HttpHeaders(); + String token = Base64Utils.encodeToString( + (this.securityProperties.getUser().getName() + ":" + this.securityProperties.getUser().getPassword()) + .getBytes(StandardCharsets.UTF_8)); + httpHeaders.set(HttpHeaders.AUTHORIZATION, "Basic " + token); + clientWebSocketContainer.setHeaders(httpHeaders); + clientWebSocketContainer.start(); + WebSocketSession session = clientWebSocketContainer.getSession(null); + session.sendMessage(new TextMessage(this.messageString)); + session.close(); + + stepVerifier.verify(); + } + + @SpringBootApplication + static class TestApplication { + } +}