iSharkFly-Open/Java-Tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

JAVA-3090: updating README.md for spring-reactive

Browse Source
This commit is contained in:
chaos2418
2021-12-29 18:42:00 +05:30
parent 0cb9125701
commit 7432cc2ee2
74 changed files with 10 additions and 3010 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
-2
View File
@@ -557,7 +557,6 @@
<module>atomikos</module>
<module>reactive-systems</module>
<module>slack</module>
<module>spring-webflux-threads</module>
</modules>
</profile>
@@ -1037,7 +1036,6 @@
<module>atomikos</module>
<module>reactive-systems</module>
<module>slack</module>
<module>spring-webflux-threads</module>
</modules>
</profile>
reactor-core/README.md
-1
View File
@@ -4,7 +4,6 @@ This module contains articles about Reactor Core.
### Relevant articles
- [Intro To Reactor Core](https://www.baeldung.com/reactor-core)
- [Combining Publishers in Project Reactor](https://www.baeldung.com/reactor-combine-streams)
- [Programmatically Creating Sequences with Project Reactor](https://www.baeldung.com/flux-sequences-reactor)
- [How to Extract a Mono’s Content in Java](https://www.baeldung.com/java-string-from-mono)
reactor-core/src/test/java/com/baeldung/reactor/introduction/ReactorIntegrationTest.java
-119
View File
@@ -1,119 +0,0 @@
package com.baeldung.reactor.introduction;
import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
public class ReactorIntegrationTest {
@Test
public void givenFlux_whenSubscribing_thenStream() {
List<Integer> elements = new ArrayList<>();
Flux.just(1, 2, 3, 4)
.log()
.map(i -> {
System.out.println(i + ":" + Thread.currentThread());
return i * 2;
})
.subscribe(elements::add);
assertThat(elements).containsExactly(2, 4, 6, 8);
}
@Test
public void givenFlux_whenZipping_thenCombine() {
List<String> elements = new ArrayList<>();
Flux.just(1, 2, 3, 4)
.log()
.map(i -> i * 2)
.zipWith(Flux.range(0, Integer.MAX_VALUE).log(), (one, two) -> String.format("First Flux: %d, Second Flux: %d", one, two))
.subscribe(elements::add);
assertThat(elements).containsExactly(
"First Flux: 2, Second Flux: 0",
"First Flux: 4, Second Flux: 1",
"First Flux: 6, Second Flux: 2",
"First Flux: 8, Second Flux: 3");
}
@Test
public void givenFlux_whenApplyingBackPressure_thenPushElementsInBatches() {
List<Integer> elements = new ArrayList<>();
Flux.just(1, 2, 3, 4)
.log()
.subscribe(new Subscriber<Integer>() {
private Subscription s;
int onNextAmount;
@Override
public void onSubscribe(final Subscription s) {
this.s = s;
s.request(2);
}
@Override
public void onNext(final Integer integer) {
elements.add(integer);
onNextAmount++;
if (onNextAmount % 2 == 0) {
s.request(2);
}
}
@Override
public void onError(final Throwable t) {
}
@Override
public void onComplete() {
}
});
assertThat(elements).containsExactly(1, 2, 3, 4);
}
@Test
public void givenFlux_whenInParallel_thenSubscribeInDifferentThreads() throws InterruptedException {
List<String> threadNames = new ArrayList<>();
Flux.just(1, 2, 3, 4)
.log()
.map(i -> Thread.currentThread().getName())
.subscribeOn(Schedulers.parallel())
.subscribe(threadNames::add);
Thread.sleep(1000);
assertThat(threadNames).containsExactly("parallel-1", "parallel-1", "parallel-1", "parallel-1");
}
@Test
public void givenConnectableFlux_whenConnected_thenShouldStream() {
List<Integer> elements = new ArrayList<>();
final ConnectableFlux<Integer> publish = Flux.just(1, 2, 3, 4).publish();
publish.subscribe(elements::add);
assertThat(elements).isEmpty();
publish.connect();
assertThat(elements).containsExactly(1, 2, 3, 4);
}
}
spring-5-reactive-2/README.md
-2
View File
@@ -2,11 +2,9 @@
This module contains articles about reactive Spring 5
- [Spring WebClient vs. RestTemplate](https://www.baeldung.com/spring-webclient-resttemplate)
- [Validation for Functional Endpoints in Spring 5](https://www.baeldung.com/spring-functional-endpoints-validation)
- [Logging a Reactive Sequence](https://www.baeldung.com/spring-reactive-sequence-logging)
- [Testing Reactive Streams Using StepVerifier and TestPublisher](https://www.baeldung.com/reactive-streams-step-verifier-test-publisher)
- [Debugging Reactive Streams in Java](https://www.baeldung.com/spring-debugging-reactive-streams)
- [Static Content in Spring WebFlux](https://www.baeldung.com/spring-webflux-static-content)
- [Server-Sent Events in Spring](https://www.baeldung.com/spring-server-sent-events)
- [Backpressure Mechanism in Spring WebFlux](https://www.baeldung.com/spring-webflux-backpressure)
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/ConsumerDebuggingApplication.java
-34
View File
@@ -1,34 +0,0 @@
package com.baeldung.debugging.consumer;
import java.util.Collections;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.mongo.MongoReactiveAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.web.server.SecurityWebFilterChain;
import reactor.core.publisher.Hooks;
@SpringBootApplication(exclude = MongoReactiveAutoConfiguration.class)
@EnableScheduling
public class ConsumerDebuggingApplication {
public static void main(String[] args) {
Hooks.onOperatorDebug();
SpringApplication app = new SpringApplication(ConsumerDebuggingApplication.class);
app.setDefaultProperties(Collections.singletonMap("server.port", "8082"));
app.run(args);
}
@Bean
public SecurityWebFilterChain debuggingConsumerSpringSecurityFilterChain(ServerHttpSecurity http) {
http.authorizeExchange()
.anyExchange()
.permitAll();
http.csrf().disable();
return http.build();
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/chronjobs/ChronJobs.java
-154
View File
@@ -1,154 +0,0 @@
package com.baeldung.debugging.consumer.chronjobs;
import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import com.baeldung.debugging.consumer.model.Foo;
import com.baeldung.debugging.consumer.model.FooDto;
import com.baeldung.debugging.consumer.service.FooService;
import reactor.core.publisher.Flux;
@Component
public class ChronJobs {
private static Logger logger = LoggerFactory.getLogger(ChronJobs.class);
private WebClient client = WebClient.create("http://localhost:8081");
@Autowired
private FooService service;
@Scheduled(fixedRate = 10000)
public void consumeInfiniteFlux() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 1 with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
Integer random = ThreadLocalRandom.current()
.nextInt(0, 3);
switch (random) {
case 0:
logger.info("process 1 with approach 1");
service.processFoo(fluxFoo);
break;
case 1:
logger.info("process 1 with approach 1 EH");
service.processUsingApproachOneWithErrorHandling(fluxFoo);
break;
default:
logger.info("process 1 with approach 2");
service.processFooInAnotherScenario(fluxFoo);
break;
}
}
@Scheduled(fixedRate = 20000)
public void consumeFiniteFlux2() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo-2")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 2 with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
Integer random = ThreadLocalRandom.current()
.nextInt(0, 3);
switch (random) {
case 0:
logger.info("process 2 with approach 1");
service.processFoo(fluxFoo);
break;
case 1:
logger.info("process 2 with approach 1 EH");
service.processUsingApproachOneWithErrorHandling(fluxFoo);
break;
default:
logger.info("process 2 with approach 2");
service.processFooInAnotherScenario(fluxFoo);
break;
}
}
@Scheduled(fixedRate = 20000)
public void consumeFiniteFlux3() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo-2")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 3 with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
logger.info("process 3 with approach 3");
service.processUsingApproachThree(fluxFoo);
}
@Scheduled(fixedRate = 20000)
public void consumeFiniteFluxWithCheckpoint4() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo-2")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 4 with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
logger.info("process 4 with approach 4");
service.processUsingApproachFourWithCheckpoint(fluxFoo);
}
@Scheduled(fixedRate = 20000)
public void consumeFiniteFluxWitParallelScheduler() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo-2")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 5-parallel with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
logger.info("process 5-parallel with approach 5-parallel");
service.processUsingApproachFivePublishingToDifferentParallelThreads(fluxFoo);
}
@Scheduled(fixedRate = 20000)
public void consumeFiniteFluxWithSingleSchedulers() {
Flux<Foo> fluxFoo = client.get()
.uri("/functional-reactive/periodic-foo-2")
.accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(FooDto.class)
.delayElements(Duration.ofMillis(100))
.map(dto -> {
logger.debug("process 5-single with dto id {} name{}", dto.getId(), dto.getName());
return new Foo(dto);
});
logger.info("process 5-single with approach 5-single");
service.processUsingApproachFivePublishingToDifferentSingleThreads(fluxFoo);
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/controllers/ReactiveConfigsToggleRestController.java
-23
View File
@@ -1,23 +0,0 @@
package com.baeldung.debugging.consumer.controllers;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Hooks;
@RestController
public class ReactiveConfigsToggleRestController {
@GetMapping("/debug-hook-on")
public String setReactiveDebugOn() {
Hooks.onOperatorDebug();
return "DEBUG HOOK ON";
}
@GetMapping("/debug-hook-off")
public String setReactiveDebugOff() {
Hooks.resetOnOperatorDebug();
return "DEBUG HOOK OFF";
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/model/Foo.java
-27
View File
@@ -1,27 +0,0 @@
package com.baeldung.debugging.consumer.model;
import java.util.concurrent.ThreadLocalRandom;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class Foo {
private Integer id;
private String formattedName;
private Integer quantity;
public Foo(FooDto dto) {
this.id = (ThreadLocalRandom.current()
.nextInt(0, 100) == 0) ? null : dto.getId();
this.formattedName = dto.getName();
this.quantity = ThreadLocalRandom.current()
.nextInt(0, 10);
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/model/FooDto.java
-16
View File
@@ -1,16 +0,0 @@
package com.baeldung.debugging.consumer.model;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
@Getter
@Setter
@NoArgsConstructor
@AllArgsConstructor
public class FooDto {
private Integer id;
private String name;
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/service/FooNameHelper.java
-45
View File
@@ -1,45 +0,0 @@
package com.baeldung.debugging.consumer.service;
import java.util.concurrent.ThreadLocalRandom;
import com.baeldung.debugging.consumer.model.Foo;
import reactor.core.publisher.Flux;
public class FooNameHelper {
public static Flux<Foo> concatAndSubstringFooName(Flux<Foo> flux) {
flux = concatFooName(flux);
flux = substringFooName(flux);
return flux;
}
public static Flux<Foo> concatFooName(Flux<Foo> flux) {
flux = flux.map(foo -> {
String processedName = null;
Integer random = ThreadLocalRandom.current()
.nextInt(0, 80);
processedName = (random != 0) ? foo.getFormattedName() : foo.getFormattedName() + "-bael";
foo.setFormattedName(processedName);
return foo;
});
return flux;
}
public static Flux<Foo> substringFooName(Flux<Foo> flux) {
return flux.map(foo -> {
String processedName;
Integer random = ThreadLocalRandom.current()
.nextInt(0, 100);
processedName = (random == 0) ? foo.getFormattedName()
.substring(10, 15)
: foo.getFormattedName()
.substring(0, 5);
foo.setFormattedName(processedName);
return foo;
});
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/service/FooQuantityHelper.java
-31
View File
@@ -1,31 +0,0 @@
package com.baeldung.debugging.consumer.service;
import java.util.concurrent.ThreadLocalRandom;
import com.baeldung.debugging.consumer.model.Foo;
import reactor.core.publisher.Flux;
public class FooQuantityHelper {
public static Flux<Foo> processFooReducingQuantity(Flux<Foo> flux) {
flux = flux.map(foo -> {
Integer result;
Integer random = ThreadLocalRandom.current()
.nextInt(0, 90);
result = (random == 0) ? result = 0 : foo.getQuantity() + 2;
foo.setQuantity(result);
return foo;
});
return divideFooQuantity(flux);
}
public static Flux<Foo> divideFooQuantity(Flux<Foo> flux) {
return flux.map(foo -> {
Integer result = Math.round(5 / foo.getQuantity());
foo.setQuantity(result);
return foo;
});
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/service/FooReporter.java
-26
View File
@@ -1,26 +0,0 @@
package com.baeldung.debugging.consumer.service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.baeldung.debugging.consumer.model.Foo;
import reactor.core.publisher.Flux;
public class FooReporter {
private static Logger logger = LoggerFactory.getLogger(FooReporter.class);
public static Flux<Foo> reportResult(Flux<Foo> input, String approach) {
return input.map(foo -> {
if (foo.getId() == null)
throw new IllegalArgumentException("Null id is not valid!");
logger.info("Reporting for approach {}: Foo with id '{}' name '{}' and quantity '{}'", approach, foo.getId(), foo.getFormattedName(), foo.getQuantity());
return foo;
});
}
public static Flux<Foo> reportResult(Flux<Foo> input) {
return reportResult(input, "default");
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/consumer/service/FooService.java
-120
View File
@@ -1,120 +0,0 @@
package com.baeldung.debugging.consumer.service;
import static com.baeldung.debugging.consumer.service.FooNameHelper.concatAndSubstringFooName;
import static com.baeldung.debugging.consumer.service.FooNameHelper.substringFooName;
import static com.baeldung.debugging.consumer.service.FooQuantityHelper.divideFooQuantity;
import static com.baeldung.debugging.consumer.service.FooQuantityHelper.processFooReducingQuantity;
import static com.baeldung.debugging.consumer.service.FooReporter.reportResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import com.baeldung.debugging.consumer.model.Foo;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
@Component
public class FooService {
private static Logger logger = LoggerFactory.getLogger(FooService.class);
public void processFoo(Flux<Foo> flux) {
flux = FooNameHelper.concatFooName(flux);
flux = FooNameHelper.substringFooName(flux);
flux = flux.log();
flux = FooReporter.reportResult(flux);
flux = flux.doOnError(error -> {
logger.error("The following error happened on processFoo method!", error);
});
flux.subscribe();
}
public void processFooInAnotherScenario(Flux<Foo> flux) {
flux = FooNameHelper.substringFooName(flux);
flux = FooQuantityHelper.divideFooQuantity(flux);
flux.subscribe();
}
public void processUsingApproachOneWithErrorHandling(Flux<Foo> flux) {
logger.info("starting approach one w error handling!");
flux = concatAndSubstringFooName(flux);
flux = concatAndSubstringFooName(flux);
flux = substringFooName(flux);
flux = processFooReducingQuantity(flux);
flux = processFooReducingQuantity(flux);
flux = processFooReducingQuantity(flux);
flux = reportResult(flux, "ONE w/ EH");
flux = flux.doOnError(error -> {
logger.error("Approach 1 with Error Handling failed!", error);
});
flux.subscribe();
}
public void processUsingApproachThree(Flux<Foo> flux) {
logger.info("starting approach three!");
flux = concatAndSubstringFooName(flux);
flux = reportResult(flux, "THREE");
flux = flux.doOnError(error -> {
logger.error("Approach 3 failed!", error);
});
flux.subscribe();
}
public void processUsingApproachFourWithCheckpoint(Flux<Foo> flux) {
logger.info("starting approach four!");
flux = concatAndSubstringFooName(flux);
flux = flux.checkpoint("CHECKPOINT 1");
flux = concatAndSubstringFooName(flux);
flux = divideFooQuantity(flux);
flux = flux.checkpoint("CHECKPOINT 2", true);
flux = reportResult(flux, "FOUR");
flux = concatAndSubstringFooName(flux).doOnError(error -> {
logger.error("Approach 4 failed!", error);
});
flux.subscribe();
}
public void processUsingApproachFourWithInitialCheckpoint(Flux<Foo> flux) {
logger.info("starting approach four!");
flux = concatAndSubstringFooName(flux);
flux = flux.checkpoint("CHECKPOINT 1", true);
flux = concatAndSubstringFooName(flux);
flux = divideFooQuantity(flux);
flux = reportResult(flux, "FOUR");
flux = flux.doOnError(error -> {
logger.error("Approach 4-2 failed!", error);
});
flux.subscribe();
}
public void processUsingApproachFivePublishingToDifferentParallelThreads(Flux<Foo> flux) {
logger.info("starting approach five-parallel!");
flux = concatAndSubstringFooName(flux).publishOn(Schedulers.newParallel("five-parallel-foo"))
.log();
flux = concatAndSubstringFooName(flux);
flux = divideFooQuantity(flux);
flux = reportResult(flux, "FIVE-PARALLEL").publishOn(Schedulers.newSingle("five-parallel-bar"));
flux = concatAndSubstringFooName(flux).doOnError(error -> {
logger.error("Approach 5-parallel failed!", error);
});
flux.subscribeOn(Schedulers.newParallel("five-parallel-starter"))
.subscribe();
}
public void processUsingApproachFivePublishingToDifferentSingleThreads(Flux<Foo> flux) {
logger.info("starting approach five-single!");
flux = flux.log()
.subscribeOn(Schedulers.newSingle("five-single-starter"));
flux = concatAndSubstringFooName(flux).publishOn(Schedulers.newSingle("five-single-foo"));
flux = concatAndSubstringFooName(flux);
flux = divideFooQuantity(flux);
flux = reportResult(flux, "FIVE-SINGLE").publishOn(Schedulers.newSingle("five-single-bar"));
flux = concatAndSubstringFooName(flux).doOnError(error -> {
logger.error("Approach 5-single failed!", error);
});
flux.subscribe();
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/server/ServerDebuggingApplication.java
-29
View File
@@ -1,29 +0,0 @@
package com.baeldung.debugging.server;
import java.util.Collections;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.web.server.SecurityWebFilterChain;
import org.springframework.web.reactive.config.EnableWebFlux;
@EnableWebFlux
@SpringBootApplication
public class ServerDebuggingApplication {
public static void main(String[] args) {
SpringApplication app = new SpringApplication(ServerDebuggingApplication.class);
app.setDefaultProperties(Collections.singletonMap("server.port", "8081"));
app.run(args);
}
@Bean
public SecurityWebFilterChain debuggingServerSpringSecurityFilterChain(ServerHttpSecurity http) {
http.authorizeExchange()
.anyExchange()
.permitAll();
return http.build();
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/server/handlers/ServerHandler.java
-47
View File
@@ -1,47 +0,0 @@
package com.baeldung.debugging.server.handlers;
import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import com.baeldung.debugging.server.model.Foo;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@Component
public class ServerHandler {
private static Logger logger = LoggerFactory.getLogger(ServerHandler.class);
public Mono<ServerResponse> useHandler(final ServerRequest request) {
// there are chances that something goes wrong here...
return ServerResponse.ok()
.contentType(MediaType.TEXT_EVENT_STREAM)
.body(Flux.interval(Duration.ofSeconds(1))
.map(sequence -> {
logger.info("retrieving Foo. Sequence: {}", sequence);
if (ThreadLocalRandom.current()
.nextInt(0, 50) == 1) {
throw new RuntimeException("There was an error retrieving the Foo!");
}
return new Foo(sequence, "name" + sequence);
}), Foo.class);
}
public Mono<ServerResponse> useHandlerFinite(final ServerRequest request) {
return ServerResponse.ok()
.contentType(MediaType.TEXT_EVENT_STREAM)
.body(Flux.range(0, 50)
.map(sequence -> {
return new Foo(new Long(sequence), "theFooNameNumber" + sequence);
}), Foo.class);
}
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/server/model/Foo.java
-13
View File
@@ -1,13 +0,0 @@
package com.baeldung.debugging.server.model;
import lombok.AllArgsConstructor;
import lombok.Data;
@Data
@AllArgsConstructor
public class Foo {
private Long id;
private String name;
}
spring-5-reactive-2/src/main/java/com/baeldung/debugging/server/routers/ServerRouter.java
-22
View File
@@ -1,22 +0,0 @@
package com.baeldung.debugging.server.routers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.RequestPredicates;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;
import com.baeldung.debugging.server.handlers.ServerHandler;
@Configuration
public class ServerRouter {
@Bean
public RouterFunction<ServerResponse> responseRoute(@Autowired ServerHandler handler) {
return RouterFunctions.route(RequestPredicates.GET("/functional-reactive/periodic-foo"), handler::useHandler)
.andRoute(RequestPredicates.GET("/functional-reactive/periodic-foo-2"), handler::useHandlerFinite);
}
}
spring-5-reactive-2/src/main/java/com/baeldung/webclient/Tweet.java
-13
View File
@@ -1,13 +0,0 @@
package com.baeldung.webclient;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Tweet {
private String text;
private String username;
}
spring-5-reactive-2/src/main/java/com/baeldung/webclient/TweetsSlowServiceController.java
-20
View File
@@ -1,20 +0,0 @@
package com.baeldung.webclient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Arrays;
import java.util.List;
@RestController
public class TweetsSlowServiceController {
@GetMapping("/slow-service-tweets")
private List<Tweet> getAllTweets() throws Exception {
Thread.sleep(2000L); // delay
return Arrays.asList(
new Tweet("RestTemplate rules", "@user1"),
new Tweet("WebClient is better", "@user2"),
new Tweet("OK, both are useful", "@user1"));
}
}
spring-5-reactive-2/src/main/java/com/baeldung/webclient/WebClientApplication.java
-23
View File
@@ -1,23 +0,0 @@
package com.baeldung.webclient;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.web.server.SecurityWebFilterChain;
@SpringBootApplication
public class WebClientApplication {
public static void main(String[] args) {
SpringApplication.run(WebClientApplication.class, args);
}
@Bean
public SecurityWebFilterChain functionalValidationsSpringSecurityFilterChain(ServerHttpSecurity http) {
http.authorizeExchange()
.anyExchange()
.permitAll();
http.csrf().disable();
return http.build();
}
}
spring-5-reactive-2/src/main/java/com/baeldung/webclient/WebController.java
-60
View File
@@ -1,60 +0,0 @@
package com.baeldung.webclient;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import java.util.List;
@Slf4j
@RestController
public class WebController {
private static final int DEFAULT_PORT = 8080;
@Setter
private int serverPort = DEFAULT_PORT;
@GetMapping("/tweets-blocking")
public List<Tweet> getTweetsBlocking() {
log.info("Starting BLOCKING Controller!");
final String uri = getSlowServiceUri();
RestTemplate restTemplate = new RestTemplate();
ResponseEntity<List<Tweet>> response = restTemplate.exchange(
uri, HttpMethod.GET, null,
new ParameterizedTypeReference<List<Tweet>>(){});
List<Tweet> result = response.getBody();
result.forEach(tweet -> log.info(tweet.toString()));
log.info("Exiting BLOCKING Controller!");
return result;
}
@GetMapping(value = "/tweets-non-blocking", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Tweet> getTweetsNonBlocking() {
log.info("Starting NON-BLOCKING Controller!");
Flux<Tweet> tweetFlux = WebClient.create()
.get()
.uri(getSlowServiceUri())
.retrieve()
.bodyToFlux(Tweet.class);
tweetFlux.subscribe(tweet -> log.info(tweet.toString()));
log.info("Exiting NON-BLOCKING Controller!");
return tweetFlux;
}
private String getSlowServiceUri() {
return "http://localhost:" + serverPort + "/slow-service-tweets";
}
}
spring-5-reactive-2/src/test/java/com/baeldung/debugging/consumer/ConsumerFooServiceIntegrationTest.java
-64
View File
@@ -1,64 +0,0 @@
package com.baeldung.debugging.consumer;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collection;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import com.baeldung.debugging.consumer.model.Foo;
import com.baeldung.debugging.consumer.service.FooService;
import com.baeldung.debugging.consumer.utils.ListAppender;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.classic.spi.IThrowableProxy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
public class ConsumerFooServiceIntegrationTest {
FooService service = new FooService();
@BeforeEach
public void clearLogList() {
Hooks.onOperatorDebug();
ListAppender.clearEventList();
}
@Test
public void givenFooWithNullId_whenProcessFoo_thenLogsWithDebugTrace() {
Foo one = new Foo(1, "nameverylong", 8);
Foo two = new Foo(null, "nameverylong", 4);
Flux<Foo> flux = Flux.just(one, two);
service.processFoo(flux);
Collection<String> allLoggedEntries = ListAppender.getEvents()
.stream()
.map(ILoggingEvent::getFormattedMessage)
.collect(Collectors.toList());
Collection<String> allSuppressedEntries = ListAppender.getEvents()
.stream()
.map(ILoggingEvent::getThrowableProxy)
.flatMap(t -> {
return Optional.ofNullable(t)
.map(IThrowableProxy::getSuppressed)
.map(Arrays::stream)
.orElse(Stream.empty());
})
.map(IThrowableProxy::getClassName)
.collect(Collectors.toList());
assertThat(allLoggedEntries).anyMatch(entry -> entry.contains("The following error happened on processFoo method!"))
.anyMatch(entry -> entry.contains("| onSubscribe"))
.anyMatch(entry -> entry.contains("| cancel()"));
assertThat(allSuppressedEntries)
.anyMatch(entry -> entry.contains("reactor.core.publisher.FluxOnAssembly$OnAssemblyException"));
}
}
spring-5-reactive-2/src/test/java/com/baeldung/debugging/consumer/ConsumerFooServiceLiveTest.java
-54
View File
@@ -1,54 +0,0 @@
package com.baeldung.debugging.consumer;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.test.web.reactive.server.WebTestClient.ResponseSpec;
import com.baeldung.debugging.consumer.service.FooService;
/**
* In order to run this live test, start the following classes:
* - com.baeldung.debugging.server.ServerDebuggingApplication
* - com.baeldung.debugging.consumer.ConsumerDebuggingApplication
*/
public class ConsumerFooServiceLiveTest {
FooService service = new FooService();
private static final String BASE_URL = "http://localhost:8082";
private static final String DEBUG_HOOK_ON = BASE_URL + "/debug-hook-on";
private static final String DEBUG_HOOK_OFF = BASE_URL + "/debug-hook-off";
private static WebTestClient client;
@BeforeAll
public static void setup() {
client = WebTestClient.bindToServer()
.baseUrl(BASE_URL)
.build();
}
@Test
public void whenRequestingDebugHookOn_thenObtainExpectedMessage() {
ResponseSpec response = client.get()
.uri(DEBUG_HOOK_ON)
.exchange();
response.expectStatus()
.isOk()
.expectBody(String.class)
.isEqualTo("DEBUG HOOK ON");
}
@Test
public void whenRequestingDebugHookOff_thenObtainExpectedMessage() {
ResponseSpec response = client.get()
.uri(DEBUG_HOOK_OFF)
.exchange();
response.expectStatus()
.isOk()
.expectBody(String.class)
.isEqualTo("DEBUG HOOK OFF");
}
}
spring-5-reactive-2/src/test/java/com/baeldung/debugging/consumer/utils/ListAppender.java
-25
View File
@@ -1,25 +0,0 @@
package com.baeldung.debugging.consumer.utils;
import java.util.ArrayList;
import java.util.List;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.AppenderBase;
public class ListAppender extends AppenderBase<ILoggingEvent> {
static private List<ILoggingEvent> events = new ArrayList<>();
@Override
protected void append(ILoggingEvent eventObject) {
events.add(eventObject);
}
public static List<ILoggingEvent> getEvents() {
return events;
}
public static void clearEventList() {
events.clear();
}
}
spring-5-reactive-2/src/test/java/com/baeldung/webclient/WebControllerIntegrationTest.java
-51
View File
@@ -1,51 +0,0 @@
package com.baeldung.webclient;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.web.reactive.server.WebTestClient;
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = WebClientApplication.class)
public class WebControllerIntegrationTest {
@LocalServerPort
int randomServerPort;
@Autowired
private WebTestClient testClient;
@Autowired
private WebController webController;
@Before
public void setup() {
webController.setServerPort(randomServerPort);
}
@Test
public void whenEndpointWithBlockingClientIsCalled_thenThreeTweetsAreReceived() {
testClient.get()
.uri("/tweets-blocking")
.exchange()
.expectStatus()
.isOk()
.expectBodyList(Tweet.class)
.hasSize(3);
}
@Test
public void whenEndpointWithNonBlockingClientIsCalled_thenThreeTweetsAreReceived() {
testClient.get()
.uri("/tweets-non-blocking")
.exchange()
.expectStatus()
.isOk()
.expectBodyList(Tweet.class)
.hasSize(3);
}
}
spring-5-reactive-security/README.md
-3
View File
@@ -8,8 +8,5 @@ The "REST With Spring" Classes: http://bit.ly/restwithspring
### Relevant Articles
- [Spring Boot Actuator](https://www.baeldung.com/spring-boot-actuators)
- [Spring Security 5 for Reactive Applications](https://www.baeldung.com/spring-security-5-reactive)
- [Guide to Spring 5 WebFlux](https://www.baeldung.com/spring-webflux)
- [Introduction to the Functional Web Framework in Spring 5](https://www.baeldung.com/spring-5-functional-web)
- [Guide to the AuthenticationManagerResolver in Spring Security](https://www.baeldung.com/spring-security-authenticationmanagerresolver)
- [Spring Webflux and CORS](https://www.baeldung.com/spring-webflux-cors)
spring-5-reactive-security/pom.xml
+1 -1
View File
@@ -110,7 +110,7 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<mainClass>com.baeldung.webflux.EmployeeSpringApplication</mainClass>
<mainClass>com.baeldung.reactive.actuator.Spring5ReactiveApplication</mainClass>
<layout>JAR</layout>
</configuration>
</plugin>
spring-5-reactive-security/src/main/java/com/baeldung/reactive/functional/EmployeeFunctionalConfig.java
-75
View File
@@ -1,75 +0,0 @@
package com.baeldung.reactive.functional;
import static org.springframework.web.reactive.function.BodyExtractors.toMono;
import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
import static org.springframework.web.reactive.function.server.ServerResponse.ok;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.web.server.SecurityWebFilterChain;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import com.baeldung.webflux.Employee;
import com.baeldung.webflux.EmployeeRepository;
@Configuration
public class EmployeeFunctionalConfig {
@Bean
EmployeeRepository employeeRepository() {
return new EmployeeRepository();
}
@Bean
RouterFunction<ServerResponse> getAllEmployeesRoute() {
return route(GET("/employees"),
req -> ok().body(
employeeRepository().findAllEmployees(), Employee.class));
}
@Bean
RouterFunction<ServerResponse> getEmployeeByIdRoute() {
return route(GET("/employees/{id}"),
req -> ok().body(
employeeRepository().findEmployeeById(req.pathVariable("id")), Employee.class));
}
@Bean
RouterFunction<ServerResponse> updateEmployeeRoute() {
return route(POST("/employees/update"),
req -> req.body(toMono(Employee.class))
.doOnNext(employeeRepository()::updateEmployee)
.then(ok().build()));
}
@Bean
RouterFunction<ServerResponse> composedRoutes() {
return
route(GET("/employees"),
req -> ok().body(
employeeRepository().findAllEmployees(), Employee.class))
.and(route(GET("/employees/{id}"),
req -> ok().body(
employeeRepository().findEmployeeById(req.pathVariable("id")), Employee.class)))
.and(route(POST("/employees/update"),
req -> req.body(toMono(Employee.class))
.doOnNext(employeeRepository()::updateEmployee)
.then(ok().build())));
}
@Bean
public SecurityWebFilterChain springSecurityFilterChain(ServerHttpSecurity http) {
http.csrf()
.disable()
.authorizeExchange()
.anyExchange()
.permitAll();
return http.build();
}
}
spring-5-reactive-security/src/main/java/com/baeldung/reactive/functional/EmployeeSpringFunctionalApplication.java
-13
View File
@@ -1,13 +0,0 @@
package com.baeldung.reactive.functional;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class EmployeeSpringFunctionalApplication {
public static void main(String[] args) {
SpringApplication.run(EmployeeSpringFunctionalApplication.class, args);
}
}
spring-5-reactive-security/src/main/java/com/baeldung/reactive/security/GreetController.java
-37
View File
@@ -1,37 +0,0 @@
package com.baeldung.reactive.security;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import java.security.Principal;
@RestController
public class GreetController {
private GreetService greetService;
public GreetController(GreetService greetService) {
this.greetService = greetService;
}
@GetMapping("/")
public Mono<String> greet(Mono<Principal> principal) {
return principal
.map(Principal::getName)
.map(name -> String.format("Hello, %s", name));
}
@GetMapping("/admin")
public Mono<String> greetAdmin(Mono<Principal> principal) {
return principal
.map(Principal::getName)
.map(name -> String.format("Admin access: %s", name));
}
@GetMapping("/greetService")
public Mono<String> greetService() {
return greetService.greet();
}
}

Some files were not shown because too many files have changed in this diff Show More