BAEL-1475: Reactive WebSockets with Spring 5
This commit is contained in:
felipeazv
parent
afb2191c4c
commit
b41ebab6b8
|
|
@ -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>Example from article: 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>
|
||||
|
|
@ -0,0 +1,11 @@
|
|||
package com.baeldung;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
@AllArgsConstructor
|
||||
public class Event {
|
||||
private String eventId;
|
||||
private String eventDt;
|
||||
}
|
||||
|
|
@ -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
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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>
|
||||
Loading…
Reference in New Issue