change module

spring-5-reactive/pom.xml

@@ -60,7 +60,13 @@
         <!-- <version>1.1.2</version> -->
         <!-- </dependency> -->
         <!-- Dependencies for Johnzon -->
-        
+        <!-- lombok-->
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<scope>compile</scope>
+		</dependency>
+		
         <dependency>
             <groupId>org.apache.geronimo.specs</groupId>
             <artifactId>geronimo-json_1.1_spec</artifactId>

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

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

spring-5-reactive/src/main/java/com/baeldung/reactive/websocket/ReactiveJavaClientWebSocket.java

@@ -0,0 +1,23 @@
+package com.baeldung.reactive.websocket;
+
+import java.net.URI;
+import java.time.Duration;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+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 ReactiveJavaClientWebSocket {
+    public static void main(String[] args) throws InterruptedException {
+        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(Duration.ofSeconds(10L));
+    }
+}

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

@@ -0,0 +1,11 @@
+package com.baeldung.reactive.websocket;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class ReactiveWebSocketApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(ReactiveWebSocketApplication.class, args);
+    }
+}

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

@@ -0,0 +1,34 @@
+package com.baeldung.reactive.websocket;
+
+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-5-reactive/src/main/java/com/baeldung/reactive/websocket/ReactiveWebSocketHandler.java

@@ -0,0 +1,71 @@
+package com.baeldung.reactive.websocket;
+
+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-5-reactive/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>