Initial commit

.gitignore

@@ -0,0 +1,25 @@
+target/
+!.mvn/wrapper/maven-wrapper.jar
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+nbproject/private/
+build/
+nbbuild/
+dist/
+nbdist/
+.nb-gradle/

pom.xml

@@ -0,0 +1,95 @@
+<?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.example</groupId>
+    <artifactId>webflux-demo</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <name>webflux-demo</name>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.0.0.RELEASE</version>
+        <relativePath/> <!-- lookup parent from repository -->
+    </parent>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+        <java.version>9</java.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                    <groupId>org.springframework.boot</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+            <exclusions>
+                <exclusion>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                    <groupId>org.springframework.boot</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-log4j2</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>23.3-jre</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-devtools</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </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>
+            </plugin>
+        </plugins>
+    </build>
+
+
+</project>

src/main/java/com/example/webfluxdemo/WebfluxDemoApplication.java

@@ -0,0 +1,77 @@
+package com.example.webfluxdemo;
+
+import com.example.webfluxdemo.model.Person;
+import com.example.webfluxdemo.repository.MongoRepository;
+import lombok.extern.slf4j.Slf4j;
+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.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import reactor.core.publisher.Mono;
+
+import java.util.Map;
+import java.util.Set;
+
+@SpringBootApplication
+@Slf4j
+public class WebfluxDemoApplication
+{
+    public static void main(String[] args)
+    {
+        SpringApplication.run(WebfluxDemoApplication.class, args);
+    }
+
+    @Bean
+    public ReactiveRedisTemplate<String, String> reactiveRedisTemplate(ReactiveRedisConnectionFactory connectionFactory)
+    {
+        return new ReactiveRedisTemplate<>(connectionFactory, RedisSerializationContext.string());
+    }
+
+    @Bean
+    public CommandLineRunner initRedis(ReactiveRedisTemplate<String, String> reactiveRedisTemplate)
+    {
+        return args ->
+        {
+            reactiveRedisTemplate.keys("*")
+                                 .doOnNext(key -> log.info("Found key in DB: " + key))
+                                 .flatMap(reactiveRedisTemplate::delete)
+                                 .doOnComplete(() -> log.info("Removed found keys."))
+                                 .then(insertIntoRedis(reactiveRedisTemplate))
+                                 .subscribe(ignored -> log.info("Thread checking message."));
+        };
+    }
+
+    @Bean
+    public CommandLineRunner savePeopleToMongo(MongoRepository mongoRepository)
+    {
+        return args ->
+        {
+            mongoRepository.deleteAll().thenMany(mongoRepository.saveAll(Set.of(
+                                           new Person(1, "Jack", 45, "Oklahoma"),
+                                           new Person(2, "John", 24, "Chicago"),
+                                           new Person(6, "Smith", 11, "London"),
+                                           new Person(5, "Rose", 39, "Kuala Lumpur"),
+                                           new Person(7, "Alexander", 44, "Amsterdam"),
+                                           new Person(3, "Jane", 32, "Brian"),
+                                           new Person(4, "Brian", 48, "New York"))))
+                           .subscribe(person -> log.info("Saved person: " + person));
+        };
+    }
+
+    private Mono<Boolean> insertIntoRedis(ReactiveRedisTemplate<String, String> reactiveRedisTemplate)
+    {
+        return reactiveRedisTemplate.opsForValue()
+                                    .multiSet(Map.of(
+                                            "one", "1",
+                                            "two", "2",
+                                            "three", "3",
+                                            "four", "4",
+                                            "six", "6",
+                                            "seven", "7",
+                                            "five", "5"))
+                                    .doOnNext(isSuccess -> log.info("Redis insert success: " + isSuccess));
+    }
+}

src/main/java/com/example/webfluxdemo/controller/PersonController.java

@@ -0,0 +1,52 @@
+package com.example.webfluxdemo.controller;
+
+import com.example.webfluxdemo.model.Person;
+import com.example.webfluxdemo.repository.MongoRepository;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+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.time.Duration;
+import java.util.Objects;
+import java.util.Set;
+
+@RestController
+public class PersonController
+{
+    private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
+    private final MongoRepository mongoRepository;
+
+    public PersonController(ReactiveRedisTemplate<String, String> reactiveRedisTemplate, MongoRepository mongoRepository)
+    {
+        this.reactiveRedisTemplate = reactiveRedisTemplate;
+        this.mongoRepository = mongoRepository;
+    }
+
+    @GetMapping("/people")
+    public Flux<Person> getPersonsByStringNumber(@RequestParam Set<String> numbers)
+    {
+        return Flux.fromIterable(numbers)
+                   .flatMap(number -> reactiveRedisTemplate.opsForValue().get(number))
+                   .filter(Objects::nonNull)
+                   .map(Integer::valueOf)
+                   .flatMap(id -> mongoRepository.findById(id))
+                   .switchIfEmpty(Mono.error(new IllegalArgumentException("Not found person.")));
+    }
+
+    @GetMapping(value = "/people-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<Person> getAllAsStream()
+    {
+        return mongoRepository.findAll().delayElements(Duration.ofMillis(1000));
+    }
+
+    @ResponseBody
+    @ExceptionHandler(IllegalArgumentException.class)
+    public ResponseEntity<?> response()
+    {
+        return ResponseEntity.status(HttpStatus.NOT_FOUND).body("Not Found");
+    }
+}

src/main/java/com/example/webfluxdemo/model/Person.java

@@ -0,0 +1,16 @@
+package com.example.webfluxdemo.model;
+
+import lombok.Data;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+@Data
+@Document
+public class Person
+{
+    @Id
+    private final int id;
+    private final String name;
+    private final int age;
+    private final String city;
+}

src/main/java/com/example/webfluxdemo/repository/MongoRepository.java

@@ -0,0 +1,10 @@
+package com.example.webfluxdemo.repository;
+
+import com.example.webfluxdemo.model.Person;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import reactor.core.publisher.Mono;
+
+public interface MongoRepository extends ReactiveCrudRepository<Person, Integer>
+{
+    Mono<Person> findById(int id);
+}

src/main/resources/application.properties

@@ -0,0 +1,2 @@
+spring.data.mongodb.host=192.168.99.100
+spring.redis.host=192.168.99.100

src/main/resources/log4j2.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN" monitorInterval="30">
+    <Properties>
+        <Property name="LOG_PATTERN">
+            %d{yyyy-MM-dd HH:mm:ss.SSS} %5p ${hostName} --- [%30.30t] %-40.40c{1.} : %m%n%ex
+        </Property>
+    </Properties>
+    <Appenders>
+        <Console name="ConsoleAppender" target="SYSTEM_OUT" follow="true">
+            <PatternLayout pattern="${LOG_PATTERN}"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="info">
+            <AppenderRef ref="ConsoleAppender"/>
+        </Root>
+    </Loggers>
+</Configuration>

src/test/java/com/example/webfluxdemo/WebfluxDemoApplicationTests.java

@@ -0,0 +1,16 @@
+package com.example.webfluxdemo;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class WebfluxDemoApplicationTests {
+
+	@Test
+	public void contextLoads() {
+	}
+
+}