diff --git a/spring-5-webflux/pom.xml b/spring-5-webflux/pom.xml index 3306fd1729..c1f537d2fe 100644 --- a/spring-5-webflux/pom.xml +++ b/spring-5-webflux/pom.xml @@ -28,12 +28,22 @@ + + 1.8 + 2.2.0.M3 + + org.springframework.boot spring-boot-starter-webflux + + org.springframework.boot + spring-boot-starter-rsocket + + org.projectlombok lombok @@ -46,8 +56,8 @@ - junit - junit + io.projectreactor + reactor-test test @@ -60,4 +70,35 @@ + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + true + + + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientApplication.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientApplication.java new file mode 100644 index 0000000000..6f35ec5ee0 --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientApplication.java @@ -0,0 +1,16 @@ +package com.baeldung.spring.rsocket.client; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; + +@SpringBootApplication +public class ClientApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder() + .main(ClientApplication.class) + .sources(ClientApplication.class) + .profiles("client") + .run(args); + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientConfiguration.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientConfiguration.java new file mode 100644 index 0000000000..7dd3591cd6 --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/ClientConfiguration.java @@ -0,0 +1,30 @@ +package com.baeldung.spring.rsocket.client; + +import io.rsocket.RSocket; +import io.rsocket.RSocketFactory; +import io.rsocket.frame.decoder.PayloadDecoder; +import io.rsocket.transport.netty.client.TcpClientTransport; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.util.MimeTypeUtils; + +@Configuration +public class ClientConfiguration { + + @Bean + public RSocket rSocket() { + return RSocketFactory.connect() + .mimeType(MimeTypeUtils.APPLICATION_JSON_VALUE, MimeTypeUtils.APPLICATION_JSON_VALUE) + .frameDecoder(PayloadDecoder.ZERO_COPY) + .transport(TcpClientTransport.create(7000)) + .start() + .block(); + } + + @Bean + RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) { + return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, rSocketStrategies); + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/MarketDataRestController.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/MarketDataRestController.java new file mode 100644 index 0000000000..3701bd69e9 --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/client/MarketDataRestController.java @@ -0,0 +1,47 @@ +package com.baeldung.spring.rsocket.client; + +import com.baeldung.spring.rsocket.model.MarketData; +import com.baeldung.spring.rsocket.model.MarketDataRequest; +import java.util.Random; +import org.reactivestreams.Publisher; +import org.springframework.http.MediaType; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RestController; + +@RestController +public class MarketDataRestController { + + private final Random random = new Random(); + private final RSocketRequester rSocketRequester; + + public MarketDataRestController(RSocketRequester rSocketRequester) { + this.rSocketRequester = rSocketRequester; + } + + @GetMapping(value = "/current/{stock}") + public Publisher current(@PathVariable("stock") String stock) { + return rSocketRequester.route("currentMarketData") + .data(new MarketDataRequest(stock)) + .retrieveMono(MarketData.class); + } + + @GetMapping(value = "/feed/{stock}", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public Publisher feed(@PathVariable("stock") String stock) { + return rSocketRequester.route("feedMarketData") + .data(new MarketDataRequest(stock)) + .retrieveFlux(MarketData.class); + } + + @GetMapping(value = "/collect") + public Publisher collect() { + return rSocketRequester.route("collectMarketData") + .data(getMarketData()) + .send(); + } + + private MarketData getMarketData() { + return new MarketData("X", random.nextInt(10)); + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketData.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketData.java new file mode 100644 index 0000000000..3eeaf449e9 --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketData.java @@ -0,0 +1,20 @@ +package com.baeldung.spring.rsocket.model; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@AllArgsConstructor +@NoArgsConstructor +public class MarketData { + + private String stock; + private int currentPrice; + + public static MarketData fromException(Exception e) { + MarketData marketData = new MarketData(); + marketData.setStock(e.getMessage()); + return marketData; + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketDataRequest.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketDataRequest.java new file mode 100644 index 0000000000..1debbb0870 --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/model/MarketDataRequest.java @@ -0,0 +1,13 @@ +package com.baeldung.spring.rsocket.model; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@AllArgsConstructor +@NoArgsConstructor +public class MarketDataRequest { + + private String stock; +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRSocketController.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRSocketController.java new file mode 100644 index 0000000000..1ff23966fa --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRSocketController.java @@ -0,0 +1,40 @@ +package com.baeldung.spring.rsocket.server; + +import com.baeldung.spring.rsocket.model.MarketData; +import com.baeldung.spring.rsocket.model.MarketDataRequest; +import org.springframework.messaging.handler.annotation.MessageExceptionHandler; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.stereotype.Controller; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +@Controller +public class MarketDataRSocketController { + + private final MarketDataRepository marketDataRepository; + + public MarketDataRSocketController(MarketDataRepository marketDataRepository) { + this.marketDataRepository = marketDataRepository; + } + + @MessageMapping("currentMarketData") + public Mono currentMarketData(MarketDataRequest marketDataRequest) { + return marketDataRepository.getOne(marketDataRequest.getStock()); + } + + @MessageMapping("feedMarketData") + public Flux feedMarketData(MarketDataRequest marketDataRequest) { + return marketDataRepository.getAll(marketDataRequest.getStock()); + } + + @MessageMapping("collectMarketData") + public Mono collectMarketData(MarketData marketData) { + marketDataRepository.add(marketData); + return Mono.empty(); + } + + @MessageExceptionHandler + public Mono handleException(Exception e) { + return Mono.just(MarketData.fromException(e)); + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRepository.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRepository.java new file mode 100644 index 0000000000..4f30c36c2f --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/MarketDataRepository.java @@ -0,0 +1,37 @@ +package com.baeldung.spring.rsocket.server; + +import com.baeldung.spring.rsocket.model.MarketData; +import java.time.Duration; +import java.util.Random; +import java.util.stream.Stream; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +@Slf4j +@Component +public class MarketDataRepository { + + private static final int BOUND = 100; + + private Random random = new Random(); + + public Flux getAll(String stock) { + return Flux.fromStream(Stream.generate(() -> getMarketDataResponse(stock))) + .log() + .delayElements(Duration.ofSeconds(1)); + } + + public Mono getOne(String stock) { + return Mono.just(getMarketDataResponse(stock)); + } + + public void add(MarketData marketData) { + log.info("New market data: {}", marketData); + } + + private MarketData getMarketDataResponse(String stock) { + return new MarketData(stock, random.nextInt(BOUND)); + } +} diff --git a/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/ServerApplication.java b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/ServerApplication.java new file mode 100644 index 0000000000..4213e315dd --- /dev/null +++ b/spring-5-webflux/src/main/java/com/baeldung/spring/rsocket/server/ServerApplication.java @@ -0,0 +1,16 @@ +package com.baeldung.spring.rsocket.server; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; + +@SpringBootApplication +public class ServerApplication { + + public static void main(String[] args) { + new SpringApplicationBuilder() + .main(ServerApplication.class) + .sources(ServerApplication.class) + .profiles("server") + .run(args); + } +} diff --git a/spring-5-webflux/src/main/resources/application-client.properties b/spring-5-webflux/src/main/resources/application-client.properties new file mode 100644 index 0000000000..4c00e40deb --- /dev/null +++ b/spring-5-webflux/src/main/resources/application-client.properties @@ -0,0 +1 @@ +server.port=8080 diff --git a/spring-5-webflux/src/main/resources/application-server.properties b/spring-5-webflux/src/main/resources/application-server.properties new file mode 100644 index 0000000000..1e343b5bf5 --- /dev/null +++ b/spring-5-webflux/src/main/resources/application-server.properties @@ -0,0 +1,2 @@ +spring.rsocket.server.port=7000 +server.port=8081 diff --git a/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/client/MarketDataRestControllerIntegrationTest.java b/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/client/MarketDataRestControllerIntegrationTest.java new file mode 100644 index 0000000000..ff00d5ec24 --- /dev/null +++ b/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/client/MarketDataRestControllerIntegrationTest.java @@ -0,0 +1,92 @@ +package com.baeldung.spring.rsocket.client; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; +import static org.springframework.http.MediaType.TEXT_EVENT_STREAM; + +import com.baeldung.spring.rsocket.model.MarketData; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketRequester.RequestSpec; +import org.springframework.messaging.rsocket.RSocketRequester.ResponseSpec; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.web.reactive.server.FluxExchangeResult; +import org.springframework.test.web.reactive.server.WebTestClient; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +@RunWith(SpringRunner.class) +@WebFluxTest(value = MarketDataRestController.class) +public class MarketDataRestControllerIntegrationTest { + + @Autowired + private WebTestClient testClient; + + @MockBean + private RSocketRequester rSocketRequester; + + @Mock + private RequestSpec requestSpec; + + @Mock + private ResponseSpec responseSpec; + + @Test + public void whenInitiatesRequest_ThenGetsResponse() throws Exception { + when(rSocketRequester.route("currentMarketData")).thenReturn(requestSpec); + when(requestSpec.data(any())).thenReturn(responseSpec); + MarketData marketData = new MarketData("X", 1); + when(responseSpec.retrieveMono(MarketData.class)).thenReturn(Mono.just(marketData)); + + testClient.get() + .uri("/current/{stock}", "X") + .exchange() + .expectStatus() + .isOk() + .expectBody(MarketData.class) + .isEqualTo(marketData); + } + + @Test + public void whenInitiatesFireAndForget_ThenGetsNoResponse() throws Exception { + when(rSocketRequester.route("collectMarketData")).thenReturn(requestSpec); + when(requestSpec.data(any())).thenReturn(responseSpec); + when(responseSpec.send()).thenReturn(Mono.empty()); + + testClient.get() + .uri("/collect") + .exchange() + .expectStatus() + .isOk() + .expectBody(Void.class); + } + + @Test + public void whenInitiatesRequest_ThenGetsStream() throws Exception { + when(rSocketRequester.route("feedMarketData")).thenReturn(requestSpec); + when(requestSpec.data(any())).thenReturn(responseSpec); + MarketData firstMarketData = new MarketData("X", 1); + MarketData secondMarketData = new MarketData("X", 2); + when(responseSpec.retrieveFlux(MarketData.class)).thenReturn(Flux.just(firstMarketData, secondMarketData)); + + FluxExchangeResult result = testClient.get() + .uri("/feed/{stock}", "X") + .accept(TEXT_EVENT_STREAM) + .exchange() + .expectStatus() + .isOk() + .returnResult(MarketData.class); + Flux marketDataFlux = result.getResponseBody(); + StepVerifier.create(marketDataFlux) + .expectNext(firstMarketData) + .expectNext(secondMarketData) + .thenCancel() + .verify(); + } +} \ No newline at end of file diff --git a/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/server/MarketDataRSocketControllerLiveTest.java b/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/server/MarketDataRSocketControllerLiveTest.java new file mode 100644 index 0000000000..dcf3b82730 --- /dev/null +++ b/spring-5-webflux/src/test/java/com/baeldung/spring/rsocket/server/MarketDataRSocketControllerLiveTest.java @@ -0,0 +1,98 @@ +package com.baeldung.spring.rsocket.server; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.verify; + +import com.baeldung.spring.rsocket.model.MarketData; +import com.baeldung.spring.rsocket.model.MarketDataRequest; +import io.rsocket.RSocket; +import io.rsocket.RSocketFactory; +import io.rsocket.frame.decoder.PayloadDecoder; +import io.rsocket.transport.netty.client.TcpClientTransport; +import java.time.Duration; +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.test.context.TestConfiguration; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Lazy; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.rsocket.RSocketStrategies; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.MimeTypeUtils; + +@RunWith(SpringRunner.class) +@SpringBootTest +@TestPropertySource(properties = {"spring.rsocket.server.port=7000"}) +public class MarketDataRSocketControllerLiveTest { + + @Autowired + private RSocketRequester rSocketRequester; + + @SpyBean + private MarketDataRSocketController rSocketController; + + @Test + public void whenGetsFireAndForget_ThenReturnsNoResponse() throws InterruptedException { + final MarketData marketData = new MarketData("X", 1); + rSocketRequester.route("collectMarketData") + .data(marketData) + .send() + .block(Duration.ofSeconds(10)); + + sleepForProcessing(); + verify(rSocketController).collectMarketData(any()); + } + + @Test + public void whenGetsRequest_ThenReturnsResponse() throws InterruptedException { + final MarketDataRequest marketDataRequest = new MarketDataRequest("X"); + rSocketRequester.route("currentMarketData") + .data(marketDataRequest) + .send() + .block(Duration.ofSeconds(10)); + + sleepForProcessing(); + verify(rSocketController).currentMarketData(any()); + } + + @Test + public void whenGetsRequest_ThenReturnsStream() throws InterruptedException { + final MarketDataRequest marketDataRequest = new MarketDataRequest("X"); + rSocketRequester.route("feedMarketData") + .data(marketDataRequest) + .send() + .block(Duration.ofSeconds(10)); + + sleepForProcessing(); + verify(rSocketController).feedMarketData(any()); + } + + private void sleepForProcessing() throws InterruptedException { + Thread.sleep(1000); + } + + @TestConfiguration + public static class ClientConfiguration { + + @Bean + @Lazy + public RSocket rSocket() { + return RSocketFactory.connect() + .mimeType(MimeTypeUtils.APPLICATION_JSON_VALUE, MimeTypeUtils.APPLICATION_JSON_VALUE) + .frameDecoder(PayloadDecoder.ZERO_COPY) + .transport(TcpClientTransport.create(7000)) + .start() + .block(); + } + + @Bean + @Lazy + RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) { + return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, rSocketStrategies); + } + } +} \ No newline at end of file diff --git a/spring-5-webflux/src/test/java/com/baeldung/spring/webclientrequests/WebClientRequestsUnitTest.java b/spring-5-webflux/src/test/java/com/baeldung/spring/webclientrequests/WebClientRequestsUnitTest.java index 8919daa9fb..353cb24d0a 100644 --- a/spring-5-webflux/src/test/java/com/baeldung/spring/webclientrequests/WebClientRequestsUnitTest.java +++ b/spring-5-webflux/src/test/java/com/baeldung/spring/webclientrequests/WebClientRequestsUnitTest.java @@ -1,5 +1,10 @@ package com.baeldung.spring.webclientrequests; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +import java.time.Duration; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -17,8 +22,6 @@ import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.util.DefaultUriBuilderFactory; import reactor.core.publisher.Mono; -import static org.mockito.Mockito.*; - @RunWith(SpringRunner.class) @WebFluxTest public class WebClientRequestsUnitTest { @@ -50,7 +53,8 @@ public class WebClientRequestsUnitTest { public void whenCallSimpleURI_thenURIMatched() { this.webClient.get() .uri("/products") - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products"); } @@ -60,7 +64,8 @@ public class WebClientRequestsUnitTest { .uri(uriBuilder -> uriBuilder .path("/products/{id}") .build(2)) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/2"); } @@ -70,7 +75,8 @@ public class WebClientRequestsUnitTest { .uri(uriBuilder -> uriBuilder .path("/products/{id}/attributes/{attributeId}") .build(2, 13)) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/2/attributes/13"); } @@ -83,7 +89,8 @@ public class WebClientRequestsUnitTest { .queryParam("color", "black") .queryParam("deliveryDate", "13/04/2019") .build()) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?name=AndroidPhone&color=black&deliveryDate=13/04/2019"); } @@ -96,7 +103,8 @@ public class WebClientRequestsUnitTest { .queryParam("color", "{authorId}") .queryParam("deliveryDate", "{date}") .build("AndroidPhone", "black", "13/04/2019")) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?name=AndroidPhone&color=black&deliveryDate=13%2F04%2F2019"); } @@ -107,7 +115,8 @@ public class WebClientRequestsUnitTest { .path("/products/") .queryParam("tag[]", "Snapdragon", "NFC") .build()) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?tag%5B%5D=Snapdragon&tag%5B%5D=NFC"); } @@ -119,7 +128,8 @@ public class WebClientRequestsUnitTest { .path("/products/") .queryParam("category", "Phones", "Tablets") .build()) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?category=Phones&category=Tablets"); } @@ -130,7 +140,8 @@ public class WebClientRequestsUnitTest { .path("/products/") .queryParam("category", String.join(",", "Phones", "Tablets")) .build()) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?category=Phones,Tablets"); } @@ -151,7 +162,8 @@ public class WebClientRequestsUnitTest { .queryParam("color", "black") .queryParam("deliveryDate", "13/04/2019") .build()) - .retrieve(); + .exchange() + .block(Duration.ofSeconds(1)); verifyCalledUrl("/products/?name=AndroidPhone&color=black&deliveryDate=13/04/2019"); }