fix(sdk): disable req resend (#1144)

By default, the request will be resent after 1 minute. This makes the timeline in chaos when not receiving data in more than 1 minute.

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

sdk/go/connection/connection.go

@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -93,6 +93,7 @@ func CreateControlChannel(pluginName string) (ControlChannel, error) {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
 	setSockOptions(sock)
+	sock.SetOption(mangos.OptionRetryTime, 0)
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
 	if err = sock.Dial(url); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())
@@ -125,6 +126,7 @@ func CreateFuncChannel(symbolName string) (DataInOutChannel, error) {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
 	setSockOptions(sock)
+	sock.SetOption(mangos.OptionRetryTime, 0)
 	url := fmt.Sprintf("ipc:///tmp/func_%s.ipc", symbolName)
 	if err = sock.Dial(url); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())

sdk/python/ekuiper/runtime/connection.py

@@ -15,13 +15,13 @@ import logging
 import time
 from typing import Callable
 
-from pynng import Req0, Push0, Pull0
+from pynng import Req0, Push0, Pull0, Timeout
 
 
 class PairChannel:
 
     def __init__(self, name: str, typ: int):
-        s = Req0()
+        s = Req0(resend_time=0)
         """TODO options"""
         if typ == 0:
             url = "ipc:///tmp/plugin_{}.ipc".format(name)
@@ -35,9 +35,13 @@ class PairChannel:
     def run(self, reply_func: Callable[[bytes], bytes]):
         self.sock.send(b'handshake')
         while True:
-            msg = self.sock.recv()
-            reply = reply_func(msg)
-            self.sock.send(reply)
+            try:
+                msg = self.sock.recv()
+                reply = reply_func(msg)
+                self.sock.send(reply)
+            except Timeout:
+                print('pair timeout')
+                pass
 
     def close(self):
         self.sock.close()