fix(portable): fail fast when plugin started and cannot dial successf…

…ully (#1198) Signed-off-by: Jiyong Huang <huangjy@emqx.io>
lf-edge · Mar 3, 2022 · 00e72c5 · 00e72c5
1 parent f8362a7
commit 00e72c5
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 4 deletions.
diff --git a/internal/plugin/portable/runtime/connection.go b/internal/plugin/portable/runtime/connection.go
@@ -71,7 +71,7 @@ func (r *NanomsgReqChannel) SendCmd(arg []byte) error {
 
 // Handshake should only be called once
 func (r *NanomsgReqChannel) Handshake() error {
-	retryCount := 3
+	retryCount := 10
 	for {
 		_, err := r.sock.Recv()
 		switch err {

diff --git a/sdk/go/connection/connection.go b/sdk/go/connection/connection.go
@@ -77,7 +77,6 @@ func (r *NanomsgRepChannel) Run(f ReplyFunc) error {
 			return fmt.Errorf("can't send reply: %s", err.Error())
 		}
 	}
-	return nil
 }
 
 func (r *NanomsgRepChannel) Close() error {
@@ -95,7 +94,9 @@ func CreateControlChannel(pluginName string) (ControlChannel, error) {
 	setSockOptions(sock)
 	sock.SetOption(mangos.OptionRetryTime, 0)
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
-	if err = sock.Dial(url); err != nil {
+	if err = sock.DialOptions(url, map[string]interface{}{
+		mangos.OptionDialAsynch: false,
+	}); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())
 	}
 	return &NanomsgRepChannel{sock: sock}, nil

diff --git a/sdk/python/ekuiper/runtime/connection.py b/sdk/python/ekuiper/runtime/connection.py
@@ -27,7 +27,11 @@ def __init__(self, name: str, typ: int):
             url = "ipc:///tmp/plugin_{}.ipc".format(name)
         else:
             url = "ipc:///tmp/func_{}.ipc".format(name)
-        s.dial(url)
+        try:
+            s.dial(url, block=True)
+        except Exception as e:
+            print(e)
+            exit(0)
         self.sock = s
 
     """ run this in a new thread"""