added sample code for postgres r2dbc

Author: Suman Das

README.md

@@ -1,18 +1,20 @@
 # Sample Reactive  - Spring Boot application
-The purpose of this project is to demonstrate how we can use [Spring WebFlux](https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html) to create a reactive web application
+The purpose of this project is to demonstrate how we can use [Spring WebFlux](https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html) to create a simple reactive web application.
+
+This project uses [PostgreSQL](https://github.com/r2dbc/r2dbc-postgresql) implementation of the R2DBC SPI.
 
 # How to build and run
 
 project can be compiled with JDK 8 and above `javac`.
 
-To compile just do `mvn clean install`.
+To compile just do `mvn clean package`.
 
 To run the application execute the following:
 ```
 java -jar target/reactive-examples*.jar
 ```
 You can also use the Swagger-ui.html to test the application.
-![alt text](swagger-ui.png)
+![alt text](react-starter-demo.png)
 
 for more detailed technical information please check my post :
 
@@ -36,7 +38,6 @@ The application contains the following REST APIs
 
 4. GET /users/events - Stream users to a browser as Server-Sent Events
 ```
-It also creates sample data in the database so that it can be tested. It is idempotent in nature.
+It contain a sample WebClient to retrieve data from our User Management application.
 
-It also contain a sample WebClient to retrieve data from our User Management application.
 

pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.2.6.RELEASE</version>
+        <version>2.3.0.RC1</version>
         <relativePath/> <!-- lookup parent from repository -->
     </parent>
     <groupId>com.reactive.examples</groupId>
@@ -21,13 +21,33 @@
     <dependencies>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-webflux</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.h2database</groupId>
+            <artifactId>h2</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.r2dbc</groupId>
+            <artifactId>r2dbc-h2</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.r2dbc</groupId>
+            <artifactId>r2dbc-postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
@@ -44,11 +64,6 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>de.flapdoodle.embed</groupId>
-            <artifactId>de.flapdoodle.embed.mongo</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-test</artifactId>
@@ -86,6 +101,18 @@
             <name>jcenter</name>
             <url>http://oss.jfrog.org/artifactory/oss-snapshot-local/</url>
         </repository>
+        <repository>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo.spring.io/milestone</url>
+        </repository>
     </repositories>
+    <pluginRepositories>
+        <pluginRepository>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo.spring.io/milestone</url>
+        </pluginRepository>
+    </pluginRepositories>
 
 </project>

react-starter-demo.png

@@ -0,0 +1,22 @@
+package com.reactive.examples.config;
+
+import io.r2dbc.spi.ConnectionFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.data.r2dbc.connectionfactory.init.CompositeDatabasePopulator;
+import org.springframework.data.r2dbc.connectionfactory.init.ConnectionFactoryInitializer;
+import org.springframework.data.r2dbc.connectionfactory.init.ResourceDatabasePopulator;
+
+@Configuration
+public class CustomConnectionFactoryInitializer {
+    @Bean
+    public ConnectionFactoryInitializer initializer(ConnectionFactory connectionFactory) {
+        ConnectionFactoryInitializer initializer = new ConnectionFactoryInitializer();
+        initializer.setConnectionFactory(connectionFactory);
+        CompositeDatabasePopulator populator = new CompositeDatabasePopulator();
+        populator.addPopulators(new ResourceDatabasePopulator(new ClassPathResource("schema.sql")));
+        initializer.setDatabasePopulator(populator);
+        return initializer;
+    }
+}

spring-boot-reactive-webflux.iml

@@ -1,16 +1,17 @@
 package com.reactive.examples.controller;
 
+import com.reactive.examples.dto.UserDepartmentDTO;
 import com.reactive.examples.model.User;
-import com.reactive.examples.model.UserCapped;
 import com.reactive.examples.service.UserService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
-import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.List;
+
 @RestController
 @RequestMapping("/users")
 public class UserController {
@@ -29,29 +30,39 @@ public Flux<User> getAllUsers(){
     }
 
     @GetMapping("/{userId}")
-    public Mono<ResponseEntity<User>> getUserById(@PathVariable String userId){
+    public Mono<ResponseEntity<User>> getUserById(@PathVariable Integer userId){
         Mono<User> user = userService.findById(userId);
         return user.map( u -> ResponseEntity.ok(u))
                 .defaultIfEmpty(ResponseEntity.notFound().build());
     }
 
     @PutMapping("/{userId}")
-    public Mono<ResponseEntity<User>> updateUserById(@PathVariable String userId, @RequestBody User user){
+    public Mono<ResponseEntity<User>> updateUserById(@PathVariable Integer userId, @RequestBody User user){
         return userService.updateUser(userId,user)
                 .map(updatedUser -> ResponseEntity.ok(updatedUser))
                 .defaultIfEmpty(ResponseEntity.badRequest().build());
     }
 
     @DeleteMapping("/{userId}")
-    public Mono<ResponseEntity<Void>> deleteUserById(@PathVariable String userId){
+    public Mono<ResponseEntity<Void>> deleteUserById(@PathVariable Integer userId){
         return userService.deleteUser(userId)
                 .map( r -> ResponseEntity.ok().<Void>build())
                 .defaultIfEmpty(ResponseEntity.notFound().build());
     }
 
-    // Users are Sent to the client as Server Sent Events
-    @GetMapping(value = "/events", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
-    public Flux<UserCapped> streamAllUsers(){
-        return userService.getStreamedUsers();
+    @GetMapping("/age/{age}")
+    public Flux<User> getUsersByAge(@PathVariable int age) {
+        return userService.findUsersByAge(age);
+    }
+
+    @PostMapping("/search/id")
+    public Flux<User> fetchUsersByIds(@RequestBody List<Integer> ids) {
+        return userService.fetchUsers(ids);
     }
+
+    @GetMapping("/{userId}/department")
+    public Mono<UserDepartmentDTO> fetchUserAndDepartment(@PathVariable Integer userId){
+        return userService.fetchUserAndDepartment(userId);
+    }
+
 }

src/main/java/com/reactive/examples/config/CustomConnectionFactoryInitializer.java

@@ -0,0 +1,20 @@
+package com.reactive.examples.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@Builder
+@NoArgsConstructor
+public class UserDepartmentDTO {
+    private Integer userId;
+    private String userName;
+    private int age;
+    private double salary;
+    private Integer departmentId;
+    private String departmentName;
+    private String loc;
+}

src/main/java/com/reactive/examples/controller/UserController.java

@@ -1,19 +1,16 @@
 package com.reactive.examples.initialize;
 
+import com.reactive.examples.model.Department;
 import com.reactive.examples.model.User;
-import com.reactive.examples.model.UserCapped;
-import com.reactive.examples.repository.UserCappedRepository;
+import com.reactive.examples.repository.DepartmentRepository;
 import com.reactive.examples.repository.UserRepository;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.context.annotation.Profile;
-import org.springframework.data.mongodb.core.CollectionOptions;
-import org.springframework.data.mongodb.core.MongoOperations;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 
-import java.time.Duration;
 import java.util.Arrays;
 import java.util.List;
 
@@ -26,35 +23,22 @@ public class UserInitializer implements CommandLineRunner {
     private UserRepository userRepository;
 
     @Autowired
-    private UserCappedRepository userCappedRepository;
-
-    @Autowired
-    private MongoOperations mongoOperations;
+    private DepartmentRepository departmentRepository;
 
     @Override
     public void run(String... args) {
             initialDataSetup();
-            createCappedCollection();
-            dataSetupForCappedCollection();
-    }
-
-    private void createCappedCollection() {
-        mongoOperations.dropCollection(UserCapped.class);
-        mongoOperations.createCollection(UserCapped.class, CollectionOptions.empty().maxDocuments(20).size(50000).capped());
     }
 
-    private void dataSetupForCappedCollection(){
-        Flux<UserCapped> userCappedFlux = Flux.interval(Duration.ofSeconds(5))
-                .map(i -> new UserCapped(null,"Stream User " + i,20,1000));
-        userCappedRepository.insert(userCappedFlux).subscribe(
-                item -> log.info("UserCapped Inserted from CommandLineRunner " + item));
-
+    private List<User> getData(){
+        return Arrays.asList(new User(null,"Suman Das",30,10000),
+                             new User(null,"Arjun Das",5,1000),
+                             new User(null,"Saurabh Ganguly",40,1000000));
     }
 
-    private List<User> getData(){
-        return Arrays.asList(new User("1","Suman Das",30,10000),
-                             new User("2","Arjun Das",5,1000),
-                             new User("3","Saurabh Ganguly",40,1000000));
+    private List<Department> getDepartments(){
+        return Arrays.asList(new Department(null,"Mechanical",1,"Mumbai"),
+                new Department(null,"Computer",2,"Bangalore"));
     }
 
     private void initialDataSetup() {
@@ -65,5 +49,15 @@ private void initialDataSetup() {
                 .subscribe(user -> {
                     log.info("User Inserted from CommandLineRunner " + user);
                 });
+
+        departmentRepository.deleteAll()
+                .thenMany(Flux.fromIterable(getDepartments()))
+                .flatMap(departmentRepository::save)
+                .thenMany(departmentRepository.findAll())
+                .subscribe(user -> {
+                    log.info("Department Inserted from CommandLineRunner " + user);
+                });
+
     }
+
 }

src/main/java/com/reactive/examples/dto/UserDepartmentDTO.java

@@ -0,0 +1,21 @@
+package com.reactive.examples.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.relational.core.mapping.Column;
+import org.springframework.data.relational.core.mapping.Table;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Table("department")
+public class Department {
+    @Id
+    private Integer id;
+    private String name;
+    @Column("user_id")
+    private Integer userId;
+    private String loc;
+}

src/main/java/com/reactive/examples/initialize/UserInitializer.java

@@ -4,15 +4,17 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.springframework.data.annotation.Id;
-import org.springframework.data.mongodb.core.mapping.Document;
+import org.springframework.data.relational.core.mapping.Table;
+
 
-@Document
 @Data
 @AllArgsConstructor
 @NoArgsConstructor
+@Table("users")
 public class User {
+
     @Id
-    private String id;
+    private Integer id;
     private String name;
     private int age;
     private double salary;

src/main/java/com/reactive/examples/model/Department.java

@@ -0,0 +1,9 @@
+package com.reactive.examples.repository;
+
+import com.reactive.examples.model.Department;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import reactor.core.publisher.Mono;
+
+public interface DepartmentRepository extends ReactiveCrudRepository<Department,Integer> {
+    Mono<Department> findByUserId(Integer userId);
+}

src/main/java/com/reactive/examples/model/User.java

@@ -1,7 +1,11 @@
 package com.reactive.examples.repository;
 
 import com.reactive.examples.model.User;
-import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.data.r2dbc.repository.Query;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import reactor.core.publisher.Flux;
 
-public interface UserRepository extends ReactiveMongoRepository<User,String> {
+public interface UserRepository extends ReactiveCrudRepository<User,Integer> {
+    @Query("select * from users where age >= $1")
+    Flux<User> findByAge(int age);
 }