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