BAEL-1814 Guide to Spring Webflux

spring-5-reactive/pom.xml

@@ -147,7 +147,15 @@
             <version>${project-reactor-test}</version>
             <scope>test</scope>
         </dependency>
-
+	    <dependency>
+	        <groupId>org.springframework.boot</groupId>
+	        <artifactId>spring-boot-starter-security</artifactId>
+	    </dependency>
+	    <dependency>
+	        <groupId>org.springframework.security</groupId>
+	        <artifactId>spring-security-test</artifactId>
+	        <scope>test</scope>
+	    </dependency>
     </dependencies>
 
     <build>

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

@@ -0,0 +1,17 @@
+package com.baeldung.reactive.webflux;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class Employee {
+
+    private String id;
+    private String name;
+
+    // standard getters and setters
+
+}

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

@@ -0,0 +1,33 @@
+package com.baeldung.reactive.webflux;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.reactive.HandlerMapping;
+import org.springframework.web.reactive.config.EnableWebFlux;
+import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
+
+@Configuration
+@EnableWebFlux
+public class EmployeeConfig {
+
+    @Bean
+    public HandlerMapping handlerMapping() {
+        Map<String, WebSocketHandler> map = new HashMap<>();
+        map.put("/employee-feed", new EmployeeWebSocketHandler());
+
+        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
+        mapping.setUrlMap(map);
+        mapping.setOrder(10);
+        return mapping;
+    }
+
+    @Bean
+    public WebSocketHandlerAdapter handlerAdapter() {
+        return new WebSocketHandlerAdapter();
+    }
+}

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

@@ -0,0 +1,38 @@
+package com.baeldung.reactive.webflux;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequestMapping("/employees")
+public class EmployeeController {
+
+    private EmployeeRepository employeeRepository;
+
+    public EmployeeController(EmployeeRepository employeeRepository) {
+        this.employeeRepository = employeeRepository;
+    }
+
+    @GetMapping("/{id}")
+    private Mono<Employee> getEmployeeById(@PathVariable String id) {
+        return employeeRepository.findEmployeeById(id);
+    }
+
+    @GetMapping
+    private Flux<Employee> getAllEmployees() {
+        return employeeRepository.findAllEmployees();
+    }
+
+    @PostMapping("/update")
+    private Mono<Employee> updateEmployee(@RequestBody Employee employee) {
+        return employeeRepository.updateEmployee(employee);
+    }
+
+}

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

@@ -0,0 +1,11 @@
+package com.baeldung.reactive.webflux;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class EmployeeCreationEvent {
+    private String employeeId;
+    private String creationTime;
+}

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

@@ -0,0 +1,64 @@
+package com.baeldung.reactive.webflux;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.stereotype.Repository;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@Repository
+public class EmployeeRepository {
+
+    static Map<String,Employee> employeeData;
+
+    static Map<String,String> employeeAccessData;
+
+    static
+    {
+        employeeData = new HashMap<>();
+        employeeData.put("1",new Employee("1","Employee 1"));
+        employeeData.put("2",new Employee("2","Employee 2"));
+        employeeData.put("3",new Employee("3","Employee 3"));
+        employeeData.put("4",new Employee("4","Employee 4"));
+        employeeData.put("5",new Employee("5","Employee 5"));
+        employeeData.put("6",new Employee("6","Employee 6"));
+        employeeData.put("7",new Employee("7","Employee 7"));
+        employeeData.put("8",new Employee("8","Employee 8"));
+        employeeData.put("9",new Employee("9","Employee 9"));
+        employeeData.put("10",new Employee("10","Employee 10"));
+
+        employeeAccessData=new HashMap<>();
+        employeeAccessData.put("1", "Employee 1 Access Key");
+        employeeAccessData.put("2", "Employee 2 Access Key");
+        employeeAccessData.put("3", "Employee 3 Access Key");
+        employeeAccessData.put("4", "Employee 4 Access Key");
+        employeeAccessData.put("5", "Employee 5 Access Key");
+        employeeAccessData.put("6", "Employee 6 Access Key");
+        employeeAccessData.put("7", "Employee 7 Access Key");
+        employeeAccessData.put("8", "Employee 8 Access Key");
+        employeeAccessData.put("9", "Employee 9 Access Key");
+        employeeAccessData.put("10", "Employee 10 Access Key");
+    }
+
+    public Mono<Employee> findEmployeeById(String id)
+    {
+        return Mono.just(employeeData.get(id));
+    }
+
+    public Flux<Employee> findAllEmployees()
+    {
+        return Flux.fromIterable(employeeData.values());
+    }
+
+    public Mono<Employee> updateEmployee(Employee employee)
+    {
+        Employee existingEmployee=employeeData.get(employee.getId());
+        if(existingEmployee!=null)
+        {
+            existingEmployee.setName(employee.getName());
+        }
+        return Mono.just(existingEmployee);
+    }
+}

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

@@ -0,0 +1,17 @@
+package com.baeldung.reactive.webflux;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class EmployeeSpringApplication {
+
+    public static void main(String[] args) {
+
+        SpringApplication.run(EmployeeSpringApplication.class, args);
+
+        EmployeeWebClient employeeWebClient = new EmployeeWebClient();
+        employeeWebClient.consume();
+    }
+
+}

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

@@ -0,0 +1,28 @@
+package com.baeldung.reactive.webflux;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class EmployeeWebClient {
+
+    WebClient client = WebClient.create("http://localhost:8080");
+
+    public void consume() {
+
+        Mono<Employee> employeeMono = client.get()
+            .uri("/employees/{id}", "1")
+            .retrieve()
+            .bodyToMono(Employee.class);
+
+        employeeMono.subscribe(System.out::println);
+
+        Flux<Employee> employeeFlux = client.get()
+            .uri("/employees")
+            .retrieve()
+            .bodyToFlux(Employee.class);
+
+        employeeFlux.subscribe(System.out::println);
+    }
+}

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

@@ -0,0 +1,38 @@
+package com.baeldung.reactive.webflux;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.HttpMethod;
+import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
+import org.springframework.security.config.web.server.ServerHttpSecurity;
+import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
+import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetails;
+import org.springframework.security.web.server.SecurityWebFilterChain;
+
+@EnableWebFluxSecurity
+public class EmployeeWebSecurityConfig {
+
+    @Bean
+    public MapReactiveUserDetailsService userDetailsService() {
+        UserDetails user = User.withDefaultPasswordEncoder()
+            .username("admin")
+            .password("password")
+            .roles("ADMIN")
+            .build();
+        return new MapReactiveUserDetailsService(user);
+    }
+
+    @Bean
+    public SecurityWebFilterChain springSecurityFilterChain(ServerHttpSecurity http) {
+        http.csrf()
+            .disable()
+            .authorizeExchange()
+            .pathMatchers(HttpMethod.POST, "/employees/update")
+            .hasRole("ADMIN")
+            .pathMatchers("/**")
+            .permitAll()
+            .and()
+            .httpBasic();
+        return http.build();
+    }
+}

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

@@ -0,0 +1,21 @@
+package com.baeldung.reactive.webflux;
+
+import java.net.URI;
+
+import org.springframework.web.reactive.socket.WebSocketMessage;
+import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
+import org.springframework.web.reactive.socket.client.WebSocketClient;
+
+public class EmployeeWebSocketClient {
+
+    public static void main(String[] args) {
+
+        WebSocketClient client = new ReactorNettyWebSocketClient();
+
+        client.execute(URI.create("ws://localhost:8080/employee-feed"), session -> session.receive()
+            .map(WebSocketMessage::getPayloadAsText)
+            .doOnNext(System.out::println)
+            .then())
+            .block(); // to subscribe and return the value
+    }
+}

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

@@ -0,0 +1,39 @@
+package com.baeldung.reactive.webflux;
+
+import static java.time.LocalDateTime.now;
+import static java.util.UUID.randomUUID;
+
+import java.time.Duration;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.WebSocketSession;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@Component
+public class EmployeeWebSocketHandler implements WebSocketHandler {
+
+    ObjectMapper om = new ObjectMapper();
+
+    @Override
+    public Mono<Void> handle(WebSocketSession webSocketSession) {
+
+        Flux<String> employeeCreationEvent = Flux.generate(sink -> {
+            EmployeeCreationEvent event = new EmployeeCreationEvent(randomUUID().toString(), now().toString());
+            try {
+                sink.next(om.writeValueAsString(event));
+            } catch (JsonProcessingException e) {
+                sink.error(e);
+            }
+        });
+
+        return webSocketSession.send(employeeCreationEvent
+            .map(webSocketSession::textMessage)
+            .delayElements(Duration.ofSeconds(1)));
+    }
+}