Merge pull request #3411 from felipeazv/master

BAEL-1475: Reactive WebSockets with Spring 5

spring-reactive-websocket/pom.xml

@@ -0,0 +1,92 @@
+<?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>spring-reactive-websocket</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>spring-reactive-websocket</name>
+	<description>Reactive WebSockets with Spring 5</description>
+
+	<parent>
+		<groupId>com.baeldung</groupId>
+		<artifactId>parent-modules</artifactId>
+		<version>1.0.0-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-integration</artifactId>
+			<version>2.0.0.M7</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+			<version>2.0.0.M7</version>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<scope>compile</scope>
+			<version>RELEASE</version>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+
+	<repositories>
+		<repository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/snapshot</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+		</repository>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/milestone</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+	</repositories>
+
+	<pluginRepositories>
+		<pluginRepository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/snapshot</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/milestone</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+	</pluginRepositories>
+
+
+</project>

spring-reactive-websocket/src/main/java/com/baeldung/Event.java

@@ -0,0 +1,11 @@
+package com.baeldung;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class Event {
+    private String eventId;
+    private String eventDt;
+}

spring-reactive-websocket/src/main/java/com/baeldung/ReactiveWebSocketApplication.java

@@ -0,0 +1,38 @@
+package com.baeldung;
+
+import java.net.URI;
+import java.time.Duration;
+
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.web.reactive.socket.WebSocketMessage;
+import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
+import org.springframework.web.reactive.socket.client.WebSocketClient;
+
+import reactor.core.publisher.Mono;
+
+@SpringBootApplication
+public class ReactiveWebSocketApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(ReactiveWebSocketApplication.class, args);
+    }
+    
+    /**
+     * Spring Reactive WebSocket Client
+     * **/
+    @Bean
+    CommandLineRunner runner() {
+        return run -> {
+            WebSocketClient client = new ReactorNettyWebSocketClient();
+            client.execute(URI.create("ws://localhost:8080/event-emitter"), session -> session.send(Mono.just(session.textMessage("event-me-from-spring-reactive-client")))
+                .thenMany(session.receive()
+                    .map(WebSocketMessage::getPayloadAsText)
+                    .log())
+                .then())
+            .block();
+//             .block(Duration.ofSeconds(10L));//force timeout after given duration
+        };
+    }
+}

spring-reactive-websocket/src/main/java/com/baeldung/ReactiveWebSocketConfiguration.java

@@ -0,0 +1,34 @@
+package com.baeldung;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.HandlerMapping;
+import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class ReactiveWebSocketConfiguration {
+
+    @Autowired
+    private WebSocketHandler webSocketHandler;
+
+    @Bean
+    public HandlerMapping webSocketHandlerMapping() {
+        Map<String, WebSocketHandler> map = new HashMap<>();
+        map.put("/event-emitter", webSocketHandler);
+
+        SimpleUrlHandlerMapping handlerMapping = new SimpleUrlHandlerMapping();
+        handlerMapping.setOrder(1);
+        handlerMapping.setUrlMap(map);
+        return handlerMapping;
+    }
+
+    @Bean
+    public WebSocketHandlerAdapter handlerAdapter() {
+        return new WebSocketHandlerAdapter();
+    }
+}

spring-reactive-websocket/src/main/java/com/baeldung/ReactiveWebSocketHandler.java

@@ -0,0 +1,71 @@
+package com.baeldung;
+
+import org.springframework.web.reactive.socket.WebSocketSession;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.WebSocketMessage;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import javax.annotation.PostConstruct;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+@Component
+public class ReactiveWebSocketHandler implements WebSocketHandler {
+
+    private Flux<Event> eventFlux;
+    private Flux<Event> intervalFlux;
+
+    /**
+     * Here we prepare a Flux that will emit a message every second
+     */
+    @PostConstruct
+    private void init() throws InterruptedException {
+
+        eventFlux = Flux.generate(e -> {
+            Event event = new Event(UUID.randomUUID()
+                .toString(),
+                LocalDateTime.now()
+                    .toString());
+            e.next(event);
+        });
+
+        intervalFlux = Flux.interval(Duration.ofMillis(1000L))
+            .zipWith(eventFlux, (time, event) -> event);
+
+    }
+
+    /**
+     * On each new client session, send the message flux to the client.
+     * Spring subscribes to the flux and send every new flux event to the WebSocketSession object
+     * @param session
+     * @return Mono<Void>
+     */
+    @Override
+    public Mono<Void> handle(WebSocketSession webSocketSession) {
+        ObjectMapper json = new ObjectMapper();
+        return webSocketSession.send(intervalFlux.map(event -> {
+            try {
+                String jsonEvent = json.writeValueAsString(event);
+                System.out.println(jsonEvent);
+                return jsonEvent;
+            } catch (JsonProcessingException e) {
+                e.printStackTrace();
+                return "";
+            }
+        })
+            .map(webSocketSession::textMessage))
+
+            .and(webSocketSession.receive()
+                .map(WebSocketMessage::getPayloadAsText)
+                .log());
+    }
+
+}

spring-reactive-websocket/src/main/resources/static/client-websocket.html

@@ -0,0 +1,34 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="UTF-8">
+<title>Baeldung: Spring 5 Reactive Client WebSocket (Browser)</title>
+</head>
+<body>
+
+	<div class="events"></div>
+	<script>
+		var clientWebSocket = new WebSocket("ws://localhost:8080/event-emitter");
+		clientWebSocket.onopen = function() {
+			console.log("clientWebSocket.onopen", clientWebSocket);
+			console.log("clientWebSocket.readyState", "websocketstatus");
+			clientWebSocket.send("event-me-from-browser");
+		}
+		clientWebSocket.onclose = function(error) {
+			console.log("clientWebSocket.onclose", clientWebSocket, error);
+			events("Closing connection");
+		}
+		clientWebSocket.onerror = function(error) {
+			console.log("clientWebSocket.onerror", clientWebSocket, error);
+			events("An error occured");
+		}
+		clientWebSocket.onmessage = function(error) {
+			console.log("clientWebSocket.onmessage", clientWebSocket, error);
+			events(error.data);
+		}
+		function events(responseEvent) {
+			document.querySelector(".events").innerHTML += responseEvent + "<br>";
+		}
+	</script>
+</body>
+</html>