ReactiveReactive Revolution — RSocket

Reference from?: https://app.pluralsight.com/library/courses/allthetalks-session-68/table-of-contents

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>2.6.4</version>
      <relativePath/> <!-- lookup parent from repository -->
   </parent>
   <groupId>com.example</groupId>
   <artifactId>service</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <name>service</name>
   <description>Demo project for Spring Boot</description>
   <properties>
      <java.version>11</java.version>
   </properties>
   <dependencies>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-data-r2dbc</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-rsocket</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-webflux</artifactId>
      </dependency>
      <dependency>
         <groupId>io.r2dbc</groupId>
         <artifactId>r2dbc-postgresql</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>
            <configuration>
               <excludes>
                  <exclude>
                     <groupId>org.projectlombok</groupId>
                     <artifactId>lombok</artifactId>
                  </exclude>
               </excludes>
            </configuration>
         </plugin>
      </plugins>
   </build>

</project>

MainApp.java

package com.example.service;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.event.EventListener;
import org.springframework.data.annotation.Id;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Controller;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.Instant;
import java.util.function.Supplier;
import java.util.stream.Stream;

import static org.springframework.web.reactive.function.server.RouterFunctions.route;

@SpringBootApplication
public class ServiceApplication {

    public static void main(String[] args) {
        SpringApplication.run(ServiceApplication.class, args);
    }


    @Bean
    public RouterFunction<ServerResponse> routes(ReservationRepository rr) {
        return route()
                .GET("/reservations", request -> ServerResponse.ok().body(rr.findAll(), Reservation.class))
                .build();
    }
}

@Component
class SampleDataInit {
    @Autowired
    private ReservationRepository reservationRepository;

    @EventListener(ApplicationReadyEvent.class)
    public void go() {
        Flux<Reservation> reservationFlux = Flux.just("Neha", "Aravind", "Laxmi", "Rajesh", "Anosh", "Sam", "Alex")
                .map(name -> Reservation.builder().name(name).build())
                .flatMap(r -> this.reservationRepository.save(r));

        this.reservationRepository.deleteAll()
                .thenMany(reservationFlux)
                .thenMany(this.reservationRepository.findAll())
                .subscribe(System.out::println);
    }
}

interface ReservationRepository extends ReactiveCrudRepository<Reservation, Integer> {

}


@AllArgsConstructor
@NoArgsConstructor
@Builder
@Data
class Reservation {
    @Id
    private Integer id;
    private String name;
}

@Controller
class GreetingService {

    @MessageMapping("greetings")
    Flux<GreetingResponse> greet(GreetingRequest request) {
        return Flux.fromStream(Stream.generate(() ->
                        new GreetingResponse("Hello " + request.getName() + " @ " + Instant.now().toString())))
                .delayElements(Duration.ofSeconds(1));
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
class GreetingRequest {
    private String name;
}

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
class GreetingResponse {
    private String message;
}

application.properties

spring.r2dbc.url=r2dbc:postgresql://localhost:5432/postgres
spring.r2dbc.username=postgres
spring.r2dbc.password=postgres
spring.rsocket.server.port=7070

init.sql

-- Drop table

-- DROP TABLE public.reservation;

CREATE TABLE public.reservation (
   id serial4 NOT NULL,
   "name" varchar NOT NULL
);

— — — — — RSocket

pom.xml

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <parent>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-parent</artifactId>
      <version>2.6.4</version>
      <relativePath/> <!-- lookup parent from repository -->
   </parent>
   <groupId>com.example</groupId>
   <artifactId>client</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <name>client</name>
   <description>Demo project for Spring Boot</description>
   <properties>
      <java.version>11</java.version>
      <spring-cloud.version>2021.0.1</spring-cloud.version>
   </properties>
   <dependencies>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-rsocket</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-webflux</artifactId>
      </dependency>
      <dependency>
         <groupId>org.springframework.cloud</groupId>
         <artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
      </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>
   <dependencyManagement>
      <dependencies>
         <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
         </dependency>
      </dependencies>
   </dependencyManagement>

   <build>
      <plugins>
         <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
               <excludes>
                  <exclude>
                     <groupId>org.projectlombok</groupId>
                     <artifactId>lombok</artifactId>
                  </exclude>
               </excludes>
            </configuration>
         </plugin>
      </plugins>
   </build>

</project>

application.properties

#server.port=9090

spring.main.web-application-type=none

MainApp.java

package com.example.client;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;

import java.io.IOException;
import java.time.Duration;

@Log4j2
@SpringBootApplication
public class ClientApplication {

    public static void main(String[] args) throws IOException {
        SpringApplication.run(ClientApplication.class, args);
        System.in.read();
    }

    @Bean
    WebClient http(WebClient.Builder builder) {
        return builder.build();
    }

    @Bean
    RSocketRequester rSocket(RSocketRequester.Builder builder) {
        return builder.tcp("localhost", 7070);
    }

    @Bean
    ApplicationRunner httpClient(WebClient http) {
        return args -> {
            http
                    .get()
                    .uri("http://localhost:8080/reservations")
                    .retrieve()
                    .bodyToFlux(Reservation.class)
                    .map(r -> r.getName())
                    .onErrorResume(ex -> Flux.just("EEKK!"))
                    .retry()
                    .timeout(Duration.ofSeconds(1))
                    .subscribe(log::info);
        };
    }

    @Bean
    public ApplicationRunner rSocketClient(RSocketRequester rSocketRequester) {
        return args -> {
            rSocketRequester.route("greetings")
                    .data(new GreetingRequest("All the Talks"))
                    .retrieveFlux(GreetingResponse.class)
                    .subscribe(log::info);
        };
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
class GreetingRequest {
    private String name;
}

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
class GreetingResponse {
    private String message;
}

@AllArgsConstructor
@NoArgsConstructor
@Builder
@Data
class Reservation {
    private Integer id;
    private String name;
}

When you run the client, please make sure your service class is also up & running.