Evaluation Article - Spring web-flux

spring-5-reactive-webflux/pom.xml

@@ -0,0 +1,60 @@
+<?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>
+
+	<groupId>com.baeldung</groupId>
+	<artifactId>spring-5-reactive-webflux</artifactId>
+	<version>1.0-SNAPSHOT</version>
+
+	<name>spring-5-reactive-webflux</name>
+	<description>A simple spring-5-reactive-webflux.</description>
+
+	<parent>
+		<groupId>com.baeldung</groupId>
+		<artifactId>parent-boot-2</artifactId>
+		<version>0.0.1-SNAPSHOT</version>
+		<relativePath>../parent-boot-2</relativePath>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-websocket</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<mainClass>com.baeldung.reactive.Spring5ReactiveApplication</mainClass>
+					<layout>JAR</layout>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+</project>

spring-5-reactive-webflux/src/main/java/com/baeldung/reactive/Spring5ReactiveApplication.java

@@ -0,0 +1,15 @@
+package com.baeldung.reactive;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * A Spring 5 reactive application
+ *
+ */
+@SpringBootApplication
+public class Spring5ReactiveApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(Spring5ReactiveApplication.class, args);
+    }
+}

spring-5-reactive-webflux/src/main/java/com/baeldung/reactive/client/CabLocationConsumer.java

@@ -0,0 +1,66 @@
+/**
+ * 
+ */
+package com.baeldung.reactive.client;
+
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.reactive.model.CabLocation;
+
+/**
+ * @author swpraman
+ *
+ */
+public class CabLocationConsumer {
+
+    private static final Logger logger = LoggerFactory.getLogger(CabLocationConsumer.class);
+
+    /**
+     * Main method to consume locations of a cab as a stream 
+     * @param args
+     */
+    public static void main(String[] args) {
+
+        // The id of the booked cab
+        String cabId = UUID.randomUUID().toString();
+
+        // URI of the API
+        String uri = "http://localhost:8080/cab/location/" + cabId;
+
+        // @formatter:off 
+        WebClient.create(uri)
+                 .get()
+                 .retrieve()
+                 .bodyToFlux(CabLocation.class)
+                 .subscribe(CabLocationConsumer::showLocation);
+        //@formatter:on 
+
+        sleepIndefinitely();
+    }
+
+    /**
+     * Helper method to print location of the cab as received from stream
+     * @param location
+     */
+    private static void showLocation(CabLocation location) {
+        logger.debug("Current Location : {}", location);
+    }
+
+    /**
+     * This method blocks the current thread indefinitely
+     */
+    private static void sleepIndefinitely() {
+        try {
+            while (true) {
+                Thread.sleep(10000);
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

spring-5-reactive-webflux/src/main/java/com/baeldung/reactive/controller/CabController.java

@@ -0,0 +1,41 @@
+/**
+ * 
+ */
+package com.baeldung.reactive.controller;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.baeldung.reactive.model.CabLocation;
+import com.baeldung.reactive.service.CabService;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author swpraman
+ *
+ */
+@RestController
+public class CabController {
+    
+    /**
+     * Logger instance 
+     */
+    private static final Logger logger = LoggerFactory.getLogger(CabController.class);
+    
+    @Autowired
+    private CabService cabService;
+    
+    // Server sends location of the cab at the interval of 1 sec
+    @GetMapping(value = "cab/location/{cabId}", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
+    public Flux<CabLocation> cabLocation(@PathVariable("cabId") String cabId) {
+            logger.debug("Getting location stream for cabId: {}", cabId);
+            return cabService.getLocation(cabId);
+    }
+
+}

spring-5-reactive-webflux/src/main/java/com/baeldung/reactive/model/CabLocation.java

@@ -0,0 +1,52 @@
+/**
+ * 
+ */
+package com.baeldung.reactive.model;
+
+import java.io.Serializable;
+
+/**
+ * @author swapanpramanick2004
+ *
+ */
+public class CabLocation implements Serializable {
+
+    /**
+     * Serial version UID
+     */
+    private static final long serialVersionUID = -3923503044822400093L;
+
+    private String cabId;
+    private double latititude;
+    private double longitude;
+
+    public String getCabId() {
+        return cabId;
+    }
+
+    public void setCabId(String cabId) {
+        this.cabId = cabId;
+    }
+
+    public double getLatititude() {
+        return latititude;
+    }
+
+    public void setLatititude(double latititude) {
+        this.latititude = latititude;
+    }
+
+    public double getLongitude() {
+        return longitude;
+    }
+
+    public void setLongitude(double longitude) {
+        this.longitude = longitude;
+    }
+
+    @Override
+    public String toString() {
+        return "CabLocation [cabId=" + cabId + ", latititude=" + latititude + ", longitude=" + longitude + "]";
+    }
+
+}

spring-5-reactive-webflux/src/main/java/com/baeldung/reactive/service/CabService.java

@@ -0,0 +1,66 @@
+/**
+ * 
+ */
+package com.baeldung.reactive.service;
+
+import java.time.Duration;
+import java.util.Random;
+import java.util.stream.Stream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import com.baeldung.reactive.model.CabLocation;
+
+import reactor.core.publisher.Flux;
+import reactor.util.function.Tuple2;
+
+/**
+ * @author swpraman
+ *
+ */
+@Service
+public class CabService {
+    
+    /**
+     * Logger instance 
+     */
+    private static final Logger logger = LoggerFactory.getLogger(CabService.class);
+    
+    /**
+     * getLocation service for cab
+     * @param cabId
+     * @return
+     */
+    public Flux<CabLocation> getLocation(String cabId) {
+        
+        // Create a flux to retrieve location
+        Flux<CabLocation> locFlux = Flux.fromStream(Stream.generate(() -> retrieveNewLocation(cabId)));
+        
+        // Zip the flux with an interval flux
+        return Flux.interval(Duration.ofSeconds(1))
+                   .zipWith(locFlux)
+                   .map(Tuple2::getT2);
+    }
+
+    /**
+     * A random instance to create random location parameters
+     */
+    private Random random = new Random();
+
+    /**
+     * A Dummy method to return random location. 
+     * In a real project it should retrieve the location from a database or any other data source.
+     * @param cabId
+     * @return
+     */
+    private CabLocation retrieveNewLocation(String cabId) {
+        logger.debug("Retrieveing location for cab: {}", cabId);
+        CabLocation location = new CabLocation();
+        location.setCabId(cabId);
+        location.setLatititude(random.nextDouble());
+        location.setLongitude(random.nextDouble());
+        return location;
+    }
+}

spring-5-reactive-webflux/src/main/resources/logback.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="reactor" level="WARN"/>
+    <logger name="com.baeldung.reactive" level="DEBUG"/>
+</configuration>

spring-5-reactive-webflux/src/site/site.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<project name="spring-5-reactive-webflux" xmlns="http://maven.apache.org/DECORATION/1.8.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/DECORATION/1.8.0 http://maven.apache.org/xsd/decoration-1.8.0.xsd">
+  <bannerLeft>
+    <name>spring-5-reactive-webflux</name>
+    <src>https://maven.apache.org/images/apache-maven-project.png</src>
+    <href>https://www.apache.org/</href>
+  </bannerLeft>
+
+  <bannerRight>
+    <src>https://maven.apache.org/images/maven-logo-black-on-white.png</src>
+    <href>https://maven.apache.org/</href>
+  </bannerRight>
+
+  <skin>
+    <groupId>org.apache.maven.skins</groupId>
+    <artifactId>maven-fluido-skin</artifactId>
+    <version>1.7</version>
+  </skin>
+
+  <body>
+    <menu ref="parent" />
+    <menu ref="reports" />
+  </body>
+</project>

spring-5-reactive-webflux/src/test/java/com/baeldung/reactive/CapLocationTests.java

@@ -0,0 +1,66 @@
+package com.baeldung.reactive;
+
+import java.time.Duration;
+import java.util.UUID;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+import com.baeldung.reactive.controller.CabController;
+import com.baeldung.reactive.model.CabLocation;
+import com.baeldung.reactive.service.CabService;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+/**
+ * Unit test for testing Cab Locations
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration(classes = { CabController.class, CabService.class })
+@WebFluxTest(controllers = { CabController.class })
+public class CapLocationTests {
+
+    @Autowired
+    private WebTestClient webClient;
+
+    @Test
+    public void whenGetAPIConsumed_thenShouldPrintTheLocationAtOneSecInterval() {
+
+        // The id of the booked cab
+        String cabId = UUID.randomUUID().toString();
+
+        // URI of the API
+        String uri = "http://localhost:8080/cab/location/" + cabId;
+
+        // @formatter:off
+        Flux<CabLocation> resultFlux = webClient.get()
+                                                .uri(uri).accept(MediaType.APPLICATION_STREAM_JSON)
+                                                .exchange()
+                                                .returnResult(CabLocation.class)
+                                                .getResponseBody();
+        
+        
+        StepVerifier.create(resultFlux)        
+                    .expectSubscription()
+                    .thenAwait(Duration.ofSeconds(1))
+                    .assertNext(location -> Assertions.assertThat(location)
+                                                      .hasFieldOrPropertyWithValue("cabId", cabId))
+                    .thenAwait(Duration.ofSeconds(1))
+                    .assertNext(location -> Assertions.assertThat(location)
+                            .hasFieldOrPropertyWithValue("cabId", cabId))
+                    .thenAwait(Duration.ofSeconds(1))
+                    .assertNext(location -> Assertions.assertThat(location)
+                            .hasFieldOrPropertyWithValue("cabId", cabId))
+                    .thenCancel()
+                    .verify();
+        
+        // @formatter:on        
+    }
+}