You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@dubbo.apache.org by "lbh6771740 (via GitHub)" <gi...@apache.org> on 2023/06/02 02:18:27 UTC

[GitHub] [dubbo] lbh6771740 opened a new issue, #12434: 使用dubbo triple reactor 方式通信，当调用方程序被杀死停止后，服务提供方建立的流doFinally方法不会收到中断信号？

lbh6771740 opened a new issue, #12434:
URL: https://github.com/apache/dubbo/issues/12434

   您好，我想咨询下使用dubbo triple reactor 方式通信的程序，当调用方程序被杀死停止后，为什么服务提供方建立的流doFinally方法不会收到中断信号？ 我理想的情况是服务提供方建立的流doFinally方法应该收到中断信号，这样我程序可以做一些资源清理操作，但目前实际程序运行并没有，希望能解答下，谢谢！
   
   环境：
   dubbo ：3.1.5
   protobuf-java: 3.22.0
   
   
   rpc.proto文件：
   ```
   syntax = "proto3";
   
   option java_multiple_files = true;
   
   package com.protocol.grpc;
   
   message Request {
     int64 id = 1;
     string sessionId = 2;
     string userId = 3;
     string ns = 4;
     string cmd = 5;
     bool binary = 6;
     bytes payload = 7;
   }
   
   enum ResponseType {
     RESPONSE = 0;
     PUSH = 1;
   }
   
   message Response {
     int64 id = 1;
     string sessionId = 2;
     string userId = 3;
     string ns = 4;
     string cmd = 5;
     bool binary = 6;
     bytes payload = 7;
     bool success = 8;
     string msg = 9;
     ResponseType type = 10;
   }
   
   service DataModelService {
     rpc connect(stream Request) returns (stream Response);
   }
   
   ```
   
   
   下面是简单的提供方代码，通过connect建立连接后，收到消息就原样返回。
   
   ```
   @Slf4j
   @DubboService(group = "test-sse")
   @RequiredArgsConstructor
   public class SseDataModelServiceProvider extends DubboDataModelServiceTriple.DataModelServiceImplBase {
   
   	@Override
   	public Flux<Response> connect(Flux<Request> request) {
   		return request.doFinally(signalType -> {
   			log.info("connect is finally, signalType:{}", signalType);
   
   			//TODO
   			//做一些资源回收
   
   		}).doOnNext(requestMsg -> {
   			log.info("get data: {}", requestMsg);
   		}).flatMap(requestMsg -> Flux.just(Response
   						.newBuilder()
   						.setId(requestMsg.getId())
   						.setSessionId(requestMsg.getSessionId())
   						.setUserId(requestMsg.getUserId())
   						.setBinary(requestMsg.getBinary())
   						.setType(ResponseType.RESPONSE)
   						.setCmd(requestMsg.getCmd())
   						.setNs(requestMsg.getNs())
   						.setSuccess(true)
   						.setPayload(ByteString.copyFrom(("payload").getBytes(StandardCharsets.UTF_8))).build())
   				);
   	}
   }
   ```
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org