[Day 30] Reactive Programming - RSocket (Hello World) Part 2

前言

补充上一篇的DEMO，包含RSC的其他应用与测试的撰写，测试是透过之前介绍过Reactor提供的StepVerifier，还没看过的人可以先去了解一下。

RSC

就用范例中的四种模式来说明，使用linux-base的朋友就麻烦再注意一下如何跳脱，这次范例是在windows环境下使用。

Request & Response

rsc tcp://localhost:8888 --route request-response --log --debug -d "{\""name\"":\""Robert\""}"
最熟悉如同http的模式。

Fire & Forget：

rsc tcp://localhost:8888 --route fire-and-forget --log --debug -d "{\""name\"":\""Robert\""}"
不会有回传。

Request Stream：

rsc tcp://localhost:8888 --stream --route greetings --log --debug -d "{\""name\"":\""Robert\""}"
透过--stream来告诉服务回传也是stream，用来区分回传是response还是stream。

Channels (Stream Request, Stream Response)

rsc tcp://localhost:8888 --channel --route stream-stream --log --debug -d -
透过-channel来判断，-d後的资料透过「-(dash)」来不断输入资料，要注意这边输入的资料会直接传送所以不需要再特别跳脱字元。
{"name":"Robert"}、{"name":"Jerry"}

Unit Test

四种互动方式的单元测试，先初始化RSocketRequester，透过builder与传入的port来建构requester。
LocalRSocketServerPort：根据文件看起来是会去抓@Value("${local.rsocket.server.port}")，推测是会自动侦测抓取本地启动的rsocket server的port，基本上就会等同於之前定义的spring.rsocket.server.port。
RSocketRequester：轻量RSocket Client，提供functional Api来建立连线、route路径与进行物件转换。

private static RSocketRequester requester; 
	@BeforeAll 
	public static void setupOnce(@Autowired RSocketRequester.Builder builder, 
			@LocalRSocketServerPort Integer port) { 
		requester = builder 
				.tcp("localhost", port); 
}

Request & Response

functional的测试可读性很高，好像没甚麽需要特别补充。

@Test
public void testRequestGetsResponse() {
 Mono<GreetingResponse> response = requester
   .route("request-response")
   .data(new GreetingRequest("Robert"))
   .retrieveMono(GreetingResponse.class);

 StepVerifier
   .create(response)
   .consumeNextWith(message -> {
    assertThat(message.getMessage()).isEqualTo("HelloRobert");
   })
   .verifyComplete();
}

FireAndForget

@Test
public void testFireAndForget() {
 Mono<Void> result = requester
   .route("fire-and-forget")
   .data(new GreetingRequest("Robert"))
   .retrieveMono(Void.class);

 StepVerifier
   .create(result)
   .verifyComplete();
}

RequestStream

RequestStream、Channel可以提前透过thenCancel()来结束。

@Test
public void testRequestStream() {
 Flux<GreetingResponse> stream = requester
   .route("greetings")
   .data(new GreetingRequest("Robert"))
   .retrieveFlux(GreetingResponse.class);

 StepVerifier
   .create(stream)
   .consumeNextWith(message -> {
    assertThat(message.getMessage()).contains("HelloRobert");
   })
   .expectNextCount(0)
   .consumeNextWith(message -> {
    assertThat(message.getMessage()).contains("HelloRobert");
   })
   .thenCancel()
   .verify();
}

Channel

@Test
public void testStreamGetsStream() {
 Flux<GreetingRequest> just = Flux
   .just(new GreetingRequest("Robert"), new GreetingRequest("Jerry"),
     new GreetingRequest("Rhys"));

 Flux<GreetingResponse> stream = requester
   .route("stream-stream")
   .data(just)
   .retrieveFlux(GreetingResponse.class);

 StepVerifier
   .create(stream)
   .consumeNextWith(message -> {
    assertThat(message.getMessage()).isEqualTo("HelloRobert");
   })
   .consumeNextWith(message -> {
    assertThat(message.getMessage()).isEqualTo("HelloJerry");
   })
   .thenCancel()
   .verify();
}

##结语
因为不想烂尾还是补了一个part 2，终於30篇结束，之後应该会整理到个人blog里面，其实还会有最後一篇心得文，让之後想参加铁人赛的人参考。
补一下之前文章的source code，但没有很完整仅供参考。

资料来源