iSharkFly-Docs
/
java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BAEL-1475: code formatting

This commit is contained in:
felipeazv 2018-01-22 22:09:57 +01:00
parent aa12596920
commit 3e5ac1de94
2 changed files with 18 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2,22 +2,25 @@ 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 {
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")))
client.execute(
URI.create("ws://localhost:8080/event-emitter"),
session -> session.send(
Mono.just(session.textMessage("event-spring-reactive-client-websocket")))
.thenMany(session.receive()
.map(WebSocketMessage::getPayloadAsText)
.log())
.map(WebSocketMessage::getPayloadAsText)
.log())
.then())
.block(Duration.ofSeconds(10L));
}
}

35
spring-5-reactive/src/main/java/com/baeldung/reactive/websocket/ReactiveWebSocketHandler.java
View File

@ -12,7 +12,6 @@ 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;
@ -20,27 +19,14 @@ import java.util.UUID;
@Component
public class ReactiveWebSocketHandler implements WebSocketHandler {
private Flux<Event> eventFlux;
private Flux<Event> intervalFlux;
private Flux<Event> eventFlux = Flux.generate(e -> {
Event event = new Event(UUID.randomUUID().toString(), LocalDateTime.now().toString());
e.next(event);
});
/**
* Here we prepare a Flux that will emit a message every second
*/
@PostConstruct
private void init() throws InterruptedException {
private Flux<Event> intervalFlux = Flux.interval(Duration.ofMillis(1000L)).zipWith(eventFlux, (time, event) -> event);
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);
}
private ObjectMapper json = new ObjectMapper();
/**
* On each new client session, send the message flux to the client.
@ -50,7 +36,7 @@ public class ReactiveWebSocketHandler implements WebSocketHandler {
*/
@Override
public Mono<Void> handle(WebSocketSession webSocketSession) {
ObjectMapper json = new ObjectMapper();
return webSocketSession.send(intervalFlux.map(event -> {
try {
String jsonEvent = json.writeValueAsString(event);
@ -60,12 +46,9 @@ public class ReactiveWebSocketHandler implements WebSocketHandler {
e.printStackTrace();
return "";
}
})
.map(webSocketSession::textMessage))
}).map(webSocketSession::textMessage))
.and(webSocketSession.receive()
.map(WebSocketMessage::getPayloadAsText)
.log());
.and(webSocketSession.receive().map(WebSocketMessage::getPayloadAsText).log());
}
}