Websocket supplier/consumer - source/sink

common/function-test-support/pom.xml

@@ -57,6 +57,10 @@
             <artifactId>spring-data-geode</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-websocket</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

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<String> 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<String> getReceivedMessages() {
+		return receivedMessages;
+	}
+}

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<Message<?>> 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.

consumer/websocket-consumer/pom.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>websocket-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>websocket-consumer</name>
+    <description>websocket consumer</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-messaging</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-websocket</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>function-test-support</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

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<Message<?>> 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<String, Object> 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);
+	}
+}

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 <tt>9292</tt>
+	 */
+	int port = DEFAULT_PORT;
+
+	/**
+	 * the number of threads for the Netty {@link io.netty.channel.EventLoopGroup}. Default is <tt>1</tt>
+	 */
+	int threads = DEFAULT_THREADS;
+
+	/**
+	 * the logLevel for netty channels. Default is <tt>WARN</tt>
+	 */
+	String logLevel = DEFAULT_LOGLEVEL;
+
+	/**
+	 * the path on which a WebsocketSink consumer needs to connect. Default is <tt>/websocket</tt>
+	 */
+	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;
+	}
+}

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 <tt>logLevel</tt>
+ * 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<Channel> channels = Collections.synchronizedList(new ArrayList<Channel>());
+
+	@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());
+	}
+
+}

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 <a href="https://bit.ly/1jVBj5T">websocket examples</a>.
+ *
+ * @author Netty Project
+ * @author Oliver Moser
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+public class WebsocketConsumerServerHandler extends SimpleChannelInboundHandler<Object> {
+
+	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<String, Object> 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;
+		}
+	}
+}

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.
+ *
+ * <ul>
+ * <li>Configure the {@link SslContext} based on {@link WebsocketConsumerProperties#ssl}</li>
+ * <li>add the {@link WebsocketConsumerServerHandler} to the underlying {@link ChannelPipeline}</li>
+ * </ul>
+ *
+ * @author Oliver Moser
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+public class WebsocketConsumerServerInitializer extends ChannelInitializer<SocketChannel> {
+
+	/**
+	 * 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;
+		}
+	}
+}

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<Trace> traces() {
+		return this.repository.findAll();
+	}
+
+	public void setEnabled(boolean enabled) {
+		this.enabled = enabled;
+	}
+
+}

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<Trace> 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<Trace> findAll() {
+		synchronized (this.traces) {
+			return Collections.unmodifiableList(this.traces);
+		}
+	}
+
+	public void add(Map<String, Object> 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);
+			}
+		}
+	}
+}

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.
+ *
+ * <p> 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<String, Object> info;
+
+	public Trace(Date timestamp, Map<String, Object> 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<String, Object> getInfo() {
+		return this.info;
+	}
+}

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<Message<?>> 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<String> 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<WebsocketConsumerClientHandler> 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<WebsocketConsumerClientHandler> handlers = createHandlerList(CLIENT_COUNT, MESSAGE_COUNT);
+
+		// submit mulitple  message
+		List<String> 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<String> submitMultipleMessages(int messageCount) {
+		List<String> 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<WebsocketConsumerClientHandler> createHandlerList(int handlerCount, int messageCount) throws
+			InterruptedException,
+			ExecutionException {
+
+		List<WebsocketConsumerClientHandler> 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 {
+
+	}
+}

pom.xml

@@ -64,6 +64,7 @@
 		<module>consumer/redis-consumer</module>
 		<module>consumer/sftp-consumer</module>
 		<module>consumer/tcp-consumer</module>
+		<module>consumer/websocket-consumer</module>
 
 		<module>function/filter-function</module>
 		<module>function/header-enricher-function</module>
@@ -83,6 +84,7 @@
 		<module>supplier/tcp-supplier</module>
 		<module>supplier/time-supplier</module>
 		<module>supplier/rabbit-supplier</module>
+		<module>supplier/websocket-supplier</module>
 
 		<module>spring-functions-parent</module>
 	</modules>

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<Flux<Message<?>>>`.
+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<Flux<Message<?>>>`.
+
+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.

supplier/websocket-supplier/pom.xml

@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>websocket-supplier</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>websocket-supplier</name>
+    <description>websocket supplier</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-websocket</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-security</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

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<Flux<Message<?>>> 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<ServerWebSocketContainer.SockJsServiceOptions> 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;
+	}
+}

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;
+		}
+
+	}
+}

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<Flux<Message<?>>> 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<Message<?>> 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 {
+	}
+}