GitHub/Spring
1
0
Fork 0
mirror of https://github.com/OtusTeam/Spring.git synced 2026-05-30 10:50:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

L21 spring-21-reactive

Browse Source
This commit is contained in:
sergey
2023-02-14 00:21:01 +03:00
parent edf00d12e4
commit 5803d6b559
17 changed files with 519 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
2022-11/spring-21-reactive/.gitignore
+24
View File
@@ -0,0 +1,24 @@
target/
### 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/
2022-11/spring-21-reactive/pom.xml
+18
View File
@@ -0,0 +1,18 @@
<?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>ru.otus</groupId>
<artifactId>spring-21-reactive</artifactId>
<version>1.0</version>
<packaging>pom</packaging>
<modules>
<module>spring-21-web-flux</module>
<module>spring-21-reactor</module>
<module>spring-21-reactive-spring-data</module>
</modules>
</project>
2022-11/spring-21-reactive/spring-21-reactive-spring-data/pom.xml
+50
View File
@@ -0,0 +1,50 @@
<?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>ru.otus</groupId>
<artifactId>spring-21-reactive-spring-data</artifactId>
<version>1.0</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.5</version>
<relativePath/>
</parent>
<properties>
<java.version>11</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
</dependency>
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongodb-driver-reactivestreams</artifactId>
</dependency>
<dependency>
<groupId>de.flapdoodle.embed</groupId>
<artifactId>de.flapdoodle.embed.mongo</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2022-11/spring-21-reactive/spring-21-reactive-spring-data/src/main/java/ru/otus/spring/ReactiveSpringDataDemo.java
+42
View File
@@ -0,0 +1,42 @@
package ru.otus.spring;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import reactor.core.publisher.Flux;
import ru.otus.spring.domain.Person;
import ru.otus.spring.repostory.PersonRepository;
import java.util.List;
@SpringBootApplication
public class ReactiveSpringDataDemo {
private static final Logger logger = LoggerFactory.getLogger(ReactiveSpringDataDemo.class);
public static void main(String[] args) throws InterruptedException {
var context = SpringApplication.run(ReactiveSpringDataDemo.class);
var repository = context.getBean(PersonRepository.class);
var persons = List.of(new Person("Pushkin"), new Person("Lermontov"));
logger.info("before save");
repository.saveAll(persons)
.doOnNext(savedPers -> logger.info("savedPers:{}", savedPers))
.subscribe();
logger.info("after save");
// можно ничего и не найти
repository.findAll()
.map(Person::getName)
.subscribe(name -> logger.info("person name:{}", name));
// Пример объединения двух Flux
Flux.merge(repository.findAll(), repository.findAll())
.map(Person::getName)
.subscribe(name -> logger.info("join name:{}", name));
Thread.sleep(60_000);
}
}
2022-11/spring-21-reactive/spring-21-reactive-spring-data/src/main/java/ru/otus/spring/domain/Person.java
+35
View File
@@ -0,0 +1,35 @@
package ru.otus.spring.domain;
public class Person {
private String id;
private String name;
public Person(String name) {
this.name = name;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "Person{" +
"id='" + id + '\'' +
", name='" + name + '\'' +
'}';
}
}
2022-11/spring-21-reactive/spring-21-reactive-spring-data/src/main/java/ru/otus/spring/repostory/PersonRepository.java
+17
View File
@@ -0,0 +1,17 @@
package ru.otus.spring.repostory;
import org.springframework.data.mongodb.repository.Query;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import ru.otus.spring.domain.Person;
public interface PersonRepository extends ReactiveMongoRepository<Person, String> {
// -ooo---ooo---ooo|---------
// --------------------X-----
Flux<Person> findByName(String name);
@Query("{ 'name': ?0 }")
Mono<Person> findFirstByName(String name);
}
2022-11/spring-21-reactive/spring-21-reactive-spring-data/src/main/resources/application.yml
+4
View File
@@ -0,0 +1,4 @@
spring:
mongodb:
embedded:
version: "3.5.5"
2022-11/spring-21-reactive/spring-21-reactor/pom.xml
+41
View File
@@ -0,0 +1,41 @@
<?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>ru.otus</groupId>
<artifactId>spring-21-reactor</artifactId>
<version>1.0</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.3</version>
<relativePath/>
</parent>
<properties>
<java.version>11</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2022-11/spring-21-reactive/spring-21-reactor/src/main/java/ru/otus/spring/Main.java
+21
View File
@@ -0,0 +1,21 @@
package ru.otus.spring;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import ru.otus.spring.reactor.ReactiveProcessingService;
@SpringBootApplication
public class Main {
public static void main(String[] args) {
var context = SpringApplication.run(Main.class);
var service = context.getBean(ReactiveProcessingService.class);
for (int i = 0; i < 10; ++i) {
service.printHello(String.format("Ivan_%d", i));
}
}
}
2022-11/spring-21-reactive/spring-21-reactor/src/main/java/ru/otus/spring/reactor/Message.java
+14
View File
@@ -0,0 +1,14 @@
package ru.otus.spring.reactor;
public class Message {
private final String value;
public Message(String value) {
this.value = value;
}
public String getValue() {
return value;
}
}
2022-11/spring-21-reactive/spring-21-reactor/src/main/java/ru/otus/spring/reactor/ReactiveProcessingService.java
+50
View File
@@ -0,0 +1,50 @@
package ru.otus.spring.reactor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.Disposable;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import ru.otus.spring.service.NonFluxService;
import javax.annotation.PreDestroy;
@Service
public class ReactiveProcessingService {
private final Logger logger = LoggerFactory.getLogger(ReactiveProcessingService.class);
private final Sinks.Many<Message> sink;
private final Disposable flow;
public ReactiveProcessingService(NonFluxService nonFluxService) {
var scheduler = Schedulers.newParallel("processing", 2);
sink = Sinks.many().unicast().onBackpressureBuffer();
flow = sink.asFlux()
.publishOn(scheduler)
.map(nonFluxService::nonFluxSayHello)
.doOnError(error -> logger.error("error", error))
.subscribe(this::printMessage);
}
/**
* Этот метод будет инициировать асинхронную обрабтку сообщения
*
* @param name это имя будет приходить из не-reactor окружения
*/
public void printHello(String name) {
var emitResult = sink.tryEmitNext(new Message(name));
logger.info("emitResult:{}", emitResult);
}
private void printMessage(Message message) {
logger.info("Message received: {}", message.getValue());
}
@PreDestroy
public void dispose() {
this.flow.dispose();
}
}
2022-11/spring-21-reactive/spring-21-reactor/src/main/java/ru/otus/spring/service/NonFluxService.java
+26
View File
@@ -0,0 +1,26 @@
package ru.otus.spring.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.otus.spring.reactor.Message;
@Service
public class NonFluxService {
private final Logger logger = LoggerFactory.getLogger(NonFluxService.class);
public Message nonFluxSayHello(Message message) {
logger.info("Message received in non-flux service: {}", message.getValue());
var name = message.getValue();
var withHello = "Hello, " + name + "!";
try {
Thread.sleep(1000);
return new Message(withHello);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
return new Message(withHello);
}
}
}
2022-11/spring-21-reactive/spring-21-web-flux/HttpRequests.http
+41
View File
@@ -0,0 +1,41 @@
###
GET http://localhost:8080/flux/one
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/flux/three
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/flux/ten
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/flux/stream
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/rx/one
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/rx/ten
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
###
GET http://localhost:8080/rx/five
Accept: */*
Content-Type: application/json
Cache-Control: no-cache
2022-11/spring-21-reactive/spring-21-web-flux/pom.xml
+45
View File
@@ -0,0 +1,45 @@
<?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>ru.otus</groupId>
<artifactId>spring-21-web-flux</artifactId>
<version>1.0</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.3</version>
<relativePath/>
</parent>
<properties>
<java.version>11</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!--
работает и без этой зависимости на Project Reactor,
но она нужна для демонстрации RxJava2 контроллера -->
<dependency>
<groupId>io.reactivex.rxjava2</groupId>
<artifactId>rxjava</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2022-11/spring-21-reactive/spring-21-web-flux/src/main/java/ru/otus/spring/ReactorController.java
+39
View File
@@ -0,0 +1,39 @@
package ru.otus.spring;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
@RestController
public class ReactorController {
@GetMapping("/flux/one")
public Mono<String> one() {
return Mono.just("one");
}
@GetMapping("/flux/three")
public Flux<Integer> three() {
return Flux.just(300, 600, 900);
}
@GetMapping("/flux/ten")
public Flux<Integer> list() {
// 12345678910|------
return Flux.range(1, 10);
}
@GetMapping(path = "/flux/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream() {
return Flux.generate(() -> 0, (state, emitter) -> {
emitter.next(state);
return state + 1;
})
.delayElements(Duration.ofSeconds(1L))
.map(Object::toString);
}
}
2022-11/spring-21-reactive/spring-21-web-flux/src/main/java/ru/otus/spring/RxJava2Controller.java
+38
View File
@@ -0,0 +1,38 @@
package ru.otus.spring;
import io.reactivex.Flowable;
import io.reactivex.Single;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.TimeUnit;
@RestController
public class RxJava2Controller {
private static final Logger logger = LoggerFactory.getLogger(RxJava2Controller.class);
@GetMapping("/rx/one")
public Single<Integer> single() {
return Single.just("one")
.map(String::length);
}
@GetMapping(value = "/rx/ten", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flowable<Long> list() {
// --0--1--2--3--4--...
return Flowable.interval(2, TimeUnit.SECONDS)
.map(i -> i + 1)
.doOnNext(item -> logger.info("item:{}", item));
}
@GetMapping(value = "/rx/five", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flowable<Long> five() {
// --0--1--2--3--4--...
return Flowable.interval(1, TimeUnit.SECONDS)
.doOnNext(item -> logger.info("item:{}", item))
.limit(5);
}
}
2022-11/spring-21-reactive/spring-21-web-flux/src/main/java/ru/otus/spring/WebFluxSpringDemo.java
+14
View File
@@ -0,0 +1,14 @@
package ru.otus.spring;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class WebFluxSpringDemo {
public static void main(String[] args) {
SpringApplication.run(WebFluxSpringDemo.class);
}
}