[Day 29] Reactive Programming - RSocket (Hello World) Part 1

前言

对RSocket有基本的了解之後，照惯例要来做一个简单的实作，内容参考Josh Long的The RSocket Revolution ，额外调整更新与说明，详细的code有放上github。(谜：第29天才开始放code)

RSocket Server

DEMO方便起见统一放在同一个档案。第一个范例是演示一个每隔一秒钟发送一次现在时间(参考The RSocket Revolution)，剩下则是实作其他三种RSocket的互动方式，最後记得要补上设定spring.rsocket.server.port，要不然不会启动。

@Controller
@Slf4j
class GreetingController {

  @MessageMapping("greetings")
  Flux<GreetingResponse> greet(GreetingRequest request) {
    log.info("greetings name:{}", request.getName());
    Stream<GreetingResponse> generate = Stream.generate(
        () -> new GreetingResponse("Hello" + request.getName() + "@" + Instant.now()));

    return Flux.fromStream(generate)
        .delayElements(Duration.ofSeconds(1)).take(10);

  }

  @MessageMapping("request-response")
  Mono<GreetingResponse> requestResponse(GreetingRequest request) {
    log.info("request-response name:{}", request.getName());
    return Mono.just(new GreetingResponse("Hello" + request.getName()));

  }
  @MessageMapping("fire-and-forget")
  Mono<Void> fireAndForget(GreetingRequest request) {
    log.info("fire-and-forget name:{}", request.getName());
    return Mono.empty();

  }
  @MessageMapping("stream-stream")
  Flux<GreetingResponse> channel(Flux<GreetingRequest> request) {
    log.info("channel....");
    return request.
        map(rq -> new GreetingResponse("Hello" +rq.getName()));

  }

}

application.properties

spring.rsocket.server.port=8888

RSocket Client

有三种方式可以呼叫RSocket的服务。

RSC

RSocket Client CLI (RSC) 是由一名日本人Toshiaki Maki开发，是专门用来呼叫RSocket的command line服务，等同於之前使用的curl。
实际呼叫上面写的服务结果如下，感觉debug模式十分强大，非常详细。

使用的过程我也真的透过这个工具找到我的问题，大部分command line工具在windows系统都会或多或少有些小问题，有跳脱了双引号，也实际echo出符合预期的指令，却怎麽样都无法通过json转换成物件，仔细一看(下图)才发现实际上传送是没有加上双引号的，最後额外补上斜线才成功。
rsc tcp://localhost:8888 --stream --route greetings --log --debug -d "{ ""name"" : ""Josh""}"

rsc tcp://localhost:8888 --stream --route greetings --log --debug -d "{\""name\"":\""Josh\""}"

RSocketRequester

因为client端不是一个服务不需一直维持启动的状态，所以补一个System.in.read()让main thread暂停，要不然会看不到执行结果就已经结束了，下面就透过ApplicationListener实际去呼叫四种不同的沟通方式，要注意channel不是用array而是Flux，在Reactive里面array也只是一个单纯的Request。

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

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

@Bean
ApplicationListener<ApplicationReadyEvent> client(RSocketRequester client) {
 return  arg -> {
  Flux<GreetingResponse> greetingResponseFlux = client.route("greetings")
    .data(new GreetingRequest("Robert"))
    .retrieveFlux(GreetingResponse.class);

  greetingResponseFlux.subscribe(System.out::println);

  Mono<GreetingResponse> requestResponse = client.route("request-response")
    .data(new GreetingRequest("request-response :Robert"))
    .retrieveMono(GreetingResponse.class);

  requestResponse.subscribe(System.out::println);

  Mono<Void> fireAndForget = client.route("fire-and-forget")
    .data(new GreetingRequest("Robert"))
    .retrieveMono(Void.class);

  fireAndForget.subscribe(System.out::println);

  Flux<GreetingResponse> channel = client.route("stream-stream")
    .data(Flux.just(new GreetingRequest("Robert"),new GreetingRequest("Jerry"),new GreetingRequest("Rhys")))
    .retrieveFlux(GreetingResponse.class);

  channel.subscribe(System.out::println);
 };
}

Spring Retrosocket

在微服务当中，少不了穿插许多呼叫其他服务的部分，若只用RestTemplate程序码相对会比较杂乱，Spring Cloud OpenFeign 提供了相对乾净整齐的介面来解决这个问题，而Spring Retrosocket就是提供给RSocket Feign风格的client端，目前还在开发实验阶段。

结语

上班日赶稿比较紧绷一点，下班时间光是准备资料都不太够，利用中午时间终於赶了一篇。RSocket还有不少内容可以研究，但今天已经是Day29，看来只能留给後人再去探索了。
注：这是一个可能没有Part2的Part1，因为下一篇是应该会是感想。

资料来源