Download as PDF, PPTX
![Client Request-Stream Server Request-Stream @GetMapping(value = ["/request-stream"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE]) fun requestStream(): Flux<Response> = requester .route("request-stream") .data(Request()) .retrieveFlux(Response::class.java) @MessageMapping("request-stream") fun requestStream(request: Request): Flux<Response> { logger.debug { "requestStream($request)" } val responseStream = Stream.generate { Response.random() } return Flux .fromStream(responseStream) .delayElements(Duration.ofSeconds(5)) }](https://image.slidesharecdn.com/reactivemicroservicesusingrsocket-190725100307/75/Reactive-micro-services-using-RSocket-35-2048.jpg)
Reactive micro services using RSocket
- 1. Reactive MicroServices using RSocket Newreactive features in Spring 5.2
- 2.
- 3. Reactive Programming isan asynchronous programming paradigm concerned with data streams and the propagation of change. - Wikipedia
- 4.
- 5.
- 6.
- 8. Binary application protocol providingreactive streams semantics using (a)sync message passing over a single connection for use on byte stream transports (as TCP, WebSockets and Aeron)
- 9.
- 10.
- 11.
- 12. 4 Symmetric InteractionModels ● Request-Response ○ Stream of one (1 to 1) ○ Mono<Payload> response = client.requestResponse(Payload request) ● Fire-and-Forget ○ No response (1 to 0) ○ Mono<Void> response = client.fireForget(Payload request) ● Request-Stream ○ Finite stream of many (1 to N) ○ Flux<Payload> out = client.requestStream(Payload request) ● Channel ○ Bi-directional streams (M to N) ○ Flux<Payload> out = client.requestChannel(Flux<Payload> in)
- 13. Transport Agnostic ● TCP ●WebSocket ● Aeron UDP
- 14. Implementations ● Java ● Kotlin ●Go ● JavaScript ● Python ● .Net ● C++
- 15. Features ● Bi-directional ● Multiplexed ●Message based ● Binary ● Protocol ● Back Pressure ● Session Resumption
- 16. Bi-Directional ● Both partnerscan be client and server ● No client/server distinction ● No requestor/responder distinction ● Even HTTP/2 makes this distinction
- 17. Multiplexed ● HTTP 1.0one connection per request => very inefficient ● HTTP 1.1 pipeline (no random order) => head-of-line blocking ● Multiple “logical streams” within one connection ● Messages are tagged with stream-id ● Random order
- 18. Message Driven BinaryProtocol ● Message split into discrete binary frames ● Machine-to-machine efficiency ● Payload can be anything
- 19. Reactive Streams BackPressure ● TCP already queues data to some extent ● Reactive Streams (pull-push) backpressure: ○ Materialize data ○ Transfer data ○ Only when receiver is ready
- 20. Session Resumption ● Long-livedstreams ● Across different transport connections ● E.g. mobile re-connection
- 21.
- 22. It removes someof the roadblocks to use reactive everywhere
- 23. Motivation ● support forinteraction models beyond request/response such as streaming responses and push ● application-level flow control semantics across network boundaries (async pull/push of bounded batch sizes) ● binary, multiplexed use of a single connection ● support resumption of long-lived subscriptions across transport connections ● need of an application protocol in order to use transport protocols such as WebSockets and Aeron
- 24. gRPC vs RSocket ●Framework ● HTTP/2 ● Protobuf ● No native web component ● No reactive semantics
- 25.
- 26. Other Incubation Projects ●Spring R2DBC ● Spring Reactive Gateway
- 27. Netifi Cloud-Native Application Platform Builton RSocket Dramatically reduces operational overhead AI-driven load balancing and management Reactive software components
- 28.
- 29.
- 30.
- 31. Client Properties ServerProperties server: port: 7001 rsocket: port: 7000 spring: rsocket: server: port: 7000 main: lazy-initialization: true <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-rsocket</artifactId> </dependency> RSocket Dependency
- 32. Client Beans @Bean fun rSocket(@Value("${rsocket.port}")port: Int) = RSocketFactory .connect() .dataMimeType(MimeTypeUtils.APPLICATION_JSON_VALUE) .frameDecoder(PayloadDecoder.ZERO_COPY) .transport(TcpClientTransport.create(port)) .start() .block() @Bean fun requester(rSocketStrategies: RSocketStrategies, @Value("${rsocket.port}") port: Int) = RSocketRequester .wrap(rSocket(port)!!, MimeTypeUtils.APPLICATION_JSON, rSocketStrategies)
- 33. Client Fire-Forget Server Fire-Forget @GetMapping("/fire-forget") funfireForget(): Mono<Void> = requester .route("fire-forget") .data(Request()) .send() @MessageMapping("fire-forget") fun fireForget(request: Request): Mono<Void> { logger.debug { "fireForget($request)" } return Mono.empty() }
- 34. Client Request-Response Server Request-Response @GetMapping("/request-response") funrequestResponse(): Mono<Response> = requester .route("request-response") .data(Request()) .retrieveMono(Response::class.java) @MessageMapping("request-response") fun requestResponse(request: Request): Mono<Response> { logger.debug { "requestResponse($request)" } val response = Response.random() return Mono.just(response) }
- 35. Client Request-Stream Server Request-Stream @GetMapping(value= ["/request-stream"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE]) fun requestStream(): Flux<Response> = requester .route("request-stream") .data(Request()) .retrieveFlux(Response::class.java) @MessageMapping("request-stream") fun requestStream(request: Request): Flux<Response> { logger.debug { "requestStream($request)" } val responseStream = Stream.generate { Response.random() } return Flux .fromStream(responseStream) .delayElements(Duration.ofSeconds(5)) }
- 36.
- 37. Links ● https://rsocket.io/ ● https://www.slideshare.net/Pivotal/welcome-to-the-reactive-revolutionrsocket-and-spring-cloud-gateway-spencer-gibb ●https://www.youtube.com/watch?v=dGNv-Djm7h0