Repository: lf-edge/ekuiper
Pull request #3181 (2024-09-09 09:24:06)

feat(conn): adapt all source/sink 

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `57.44681%` with `60 lines` in your changes missing coverage. Please review.
> Project coverage is 71.07%. Comparing base [(`79cd2d6`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/79cd2d6c7779a8c186d164a6eba3f1caa12f6cac?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`d60dbf1`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/d60dbf12e8c31761e09c4ce0ac75755de222109a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files with missing lines](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [extensions/impl/influx2/influx2.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=extensions%2Fimpl%2Finflux2%2Finflux2.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL2luZmx1eDIvaW5mbHV4Mi5nbw==) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/impl/zmq/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=extensions%2Fimpl%2Fzmq%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL3ptcS9zaW5rLmdv) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/impl/zmq/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=extensions%2Fimpl%2Fzmq%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL3ptcS9zb3VyY2UuZ28=) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/impl/influx/influx.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=extensions%2Fimpl%2Finflux%2Finflux.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL2luZmx1eC9pbmZsdXguZ28=) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/memory/lookupsource.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=internal%2Fio%2Fmemory%2Flookupsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vbWVtb3J5L2xvb2t1cHNvdXJjZS5nbw==) | 33.33% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/lookup.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2Flookup.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvbG9va3VwLmdv) | 42.86% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/redisPub.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2FredisPub.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvcmVkaXNQdWIuZ28=) | 42.86% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/redisSub.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2FredisSub.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvcmVkaXNTdWIuZ28=) | 42.86% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/redis/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=internal%2Fio%2Fredis%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vcmVkaXMvc2luay5nbw==) | 42.86% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [extensions/impl/kafka/sink.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&filepath=extensions%2Fimpl%2Fkafka%2Fsink.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-ZXh0ZW5zaW9ucy9pbXBsL2thZmthL3NpbmsuZ28=) | 62.50% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| ... and [4 more](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3181      +/-   ##
==========================================
- Coverage   71.16%   71.07%   -0.09%     
==========================================
  Files         386      386              
  Lines       43290    43384      +94     
==========================================
+ Hits        30806    30834      +28     
- Misses      10051    10108      +57     
- Partials     2433     2442       +9     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3181?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/extensions/impl/image/image.go b/extensions/impl/image/image.go
index a6551d1f..265604d0 100644
--- a/extensions/impl/image/image.go
+++ b/extensions/impl/image/image.go
@@ -72,6 +72,7 @@ func (m *imageSink) Provision(_ api.StreamContext, configs map[string]any) error
 func (m *imageSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	if _, err := os.Stat(m.c.Path); os.IsNotExist(err) {
 		if err := os.MkdirAll(m.c.Path, os.ModePerm); nil != err {
+			sch(api.ConnectionDisconnected, err.Error())
 			return fmt.Errorf("fail to open image sink for %v", err)
 		}
 	}
diff --git a/extensions/impl/influx/influx.go b/extensions/impl/influx/influx.go
index fce82c66..ac35cf35 100644
--- a/extensions/impl/influx/influx.go
+++ b/extensions/impl/influx/influx.go
@@ -89,6 +89,13 @@ func (m *influxSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler)
 	if m.tlsconf != nil {
 		insecureSkip = m.tlsconf.InsecureSkipVerify
 	}
+	defer func() {
+		if err != nil {
+			sch(api.ConnectionDisconnected, err.Error())
+		} else {
+			sch(api.ConnectionConnecting, "")
+		}
+	}()
 	m.cli, err = client.NewHTTPClient(client.HTTPConfig{
 		Addr:               m.conf.Addr,
 		Username:           m.conf.Username,
diff --git a/extensions/impl/influx2/influx2.go b/extensions/impl/influx2/influx2.go
index 603751ab..fba53fa8 100644
--- a/extensions/impl/influx2/influx2.go
+++ b/extensions/impl/influx2/influx2.go
@@ -133,14 +133,21 @@ func (m *influxSink2) Provision(_ api.StreamContext, props map[string]any) error
 	return nil
 }
 
-func (m *influxSink2) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+func (m *influxSink2) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) (err error) {
 	options := client.DefaultOptions().SetPrecision(m.conf.Precision).SetBatchSize(uint(m.conf.BatchSize))
 	if m.tlsconf != nil {
 		options = options.SetTLSConfig(m.tlsconf)
 	}
+	defer func() {
+		if err != nil {
+			sch(api.ConnectionDisconnected, err.Error())
+		} else {
+			sch(api.ConnectionConnecting, "")
+		}
+	}()
 	m.cli = client.NewClientWithOptions(m.conf.Addr, m.conf.Token, options)
 	// Test connection
-	_, err := m.cli.Ping(ctx)
+	_, err = m.cli.Ping(ctx)
 	return err
 }
 
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
index 719b5237..5e25da48 100644
--- a/extensions/impl/kafka/sink.go
+++ b/extensions/impl/kafka/sink.go
@@ -131,8 +131,14 @@ func (k *KafkaSink) Close(ctx api.StreamContext) error {
 	return k.writer.Close()
 }
 
-func (k *KafkaSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
-	return k.buildKafkaWriter()
+func (k *KafkaSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	err := k.buildKafkaWriter()
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+	} else {
+		sch(api.ConnectionConnecting, "")
+	}
+	return err
 }
 
 func (k *KafkaSink) Collect(ctx api.StreamContext, item api.MessageTuple) error {
diff --git a/extensions/impl/kafka/source.go b/extensions/impl/kafka/source.go
index 6d1a6135..ed0b900e 100644
--- a/extensions/impl/kafka/source.go
+++ b/extensions/impl/kafka/source.go
@@ -134,7 +134,7 @@ func (k *KafkaSource) Close(ctx api.StreamContext) error {
 	return k.reader.Close()
 }
 
-func (k *KafkaSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (k *KafkaSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	readerConfig := k.sc.GetReaderConfig()
 	conf.Log.Infof("topic: %s, brokers: %v", readerConfig.Topic, readerConfig.Brokers)
 	readerConfig.Dialer = &kafkago.Dialer{
@@ -149,6 +149,11 @@ func (k *KafkaSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler)
 	if err != nil {
 		return err
 	}
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+	} else {
+		sch(api.ConnectionConnecting, "")
+	}
 	return nil
 }
 
diff --git a/extensions/impl/random/random.go b/extensions/impl/random/random.go
index 078a83f2..cbcd878c 100644
--- a/extensions/impl/random/random.go
+++ b/extensions/impl/random/random.go
@@ -65,7 +65,7 @@ func (s *randomSource) Provision(ctx api.StreamContext, props map[string]any) er
 	return nil
 }
 
-func (s *randomSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *randomSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	logger := ctx.GetLogger()
 	logger.Debugf("open random source with deduplicate %d", s.conf.Deduplicate)
 	if s.conf.Deduplicate != 0 {
@@ -85,6 +85,7 @@ func (s *randomSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler)
 			}
 		}
 	}
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/extensions/impl/video/source.go b/extensions/impl/video/source.go
index 7c5870ed..ed90f257 100644
--- a/extensions/impl/video/source.go
+++ b/extensions/impl/video/source.go
@@ -77,7 +77,8 @@ func (s *Source) Close(_ api.StreamContext) error {
 	return nil
 }
 
-func (s *Source) Connect(_ api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *Source) Connect(_ api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/extensions/impl/zmq/sink.go b/extensions/impl/zmq/sink.go
index 7b683fa0..62334740 100644
--- a/extensions/impl/zmq/sink.go
+++ b/extensions/impl/zmq/sink.go
@@ -39,7 +39,14 @@ func (m *zmqSink) Provision(ctx api.StreamContext, configs map[string]any) error
 	return nil
 }
 
-func (m *zmqSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) (err error) {
+func (m *zmqSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) (err error) {
+	defer func() {
+		if err != nil {
+			sch(api.ConnectionDisconnected, err.Error())
+		} else {
+			sch(api.ConnectionConnecting, "")
+		}
+	}()
 	m.publisher, err = zmq.NewSocket(zmq.PUB)
 	if err != nil {
 		return fmt.Errorf("zmq sink fails to create socket: %v", err)
diff --git a/extensions/impl/zmq/source.go b/extensions/impl/zmq/source.go
index 3ecd2fbd..9a1cfcfc 100644
--- a/extensions/impl/zmq/source.go
+++ b/extensions/impl/zmq/source.go
@@ -39,8 +39,15 @@ func (s *zmqSource) Provision(ctx api.StreamContext, configs map[string]any) err
 	return nil
 }
 
-func (s *zmqSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *zmqSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	var err error
+	defer func() {
+		if err != nil {
+			sch(api.ConnectionDisconnected, err.Error())
+		} else {
+			sch(api.ConnectionConnecting, "")
+		}
+	}()
 	s.subscriber, err = zmq.NewSocket(zmq.SUB)
 	if err != nil {
 		return fmt.Errorf("zmq source fails to create socket: %v", err)
diff --git a/internal/io/file/sink.go b/internal/io/file/sink.go
index fdc18554..67cf3a26 100644
--- a/internal/io/file/sink.go
+++ b/internal/io/file/sink.go
@@ -121,7 +121,7 @@ func (m *fileSink) Provision(ctx api.StreamContext, props map[string]interface{}
 	return nil
 }
 
-func (m *fileSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (m *fileSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	ctx.GetLogger().Debug("Opening file sink")
 	// Check if the files have opened longer than the rolling interval, if so close it and create a new one
 	if m.c.CheckInterval > 0 {
@@ -155,6 +155,7 @@ func (m *fileSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) err
 			}
 		}()
 	}
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 0e50232e..26db5c69 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -143,8 +143,8 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 	return nil
 }
 
-func (fs *Source) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
-	// do nothing
+func (fs *Source) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index 72f459fb..8e298950 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -41,7 +41,8 @@ func (hps *HttpPullSource) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (hps *HttpPullSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (hps *HttpPullSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/http/httppush_source.go b/internal/io/http/httppush_source.go
index 2899ceb2..5de5664f 100644
--- a/internal/io/http/httppush_source.go
+++ b/internal/io/http/httppush_source.go
@@ -72,8 +72,8 @@ func (h *HttpPushSource) Close(ctx api.StreamContext) error {
 	return connection.DetachConnection(ctx, h.conf.DataSource)
 }
 
-func (h *HttpPushSource) Connect(ctx api.StreamContext, sc api.StatusChangeHandler) error {
-	cw, err := connection.FetchConnection(ctx, h.conf.DataSource, "httppush", h.props, sc)
+func (h *HttpPushSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	cw, err := connection.FetchConnection(ctx, h.conf.DataSource, "httppush", h.props, sch)
 	if err != nil {
 		return err
 	}
@@ -87,6 +87,7 @@ func (h *HttpPushSource) Connect(ctx api.StreamContext, sc api.StatusChangeHandl
 	}
 	h.sourceID = fmt.Sprintf("%s_%s_%v", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	h.topic = hc.GetTopic()
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/http/lookup_source.go b/internal/io/http/lookup_source.go
index df748802..f8781989 100644
--- a/internal/io/http/lookup_source.go
+++ b/internal/io/http/lookup_source.go
@@ -39,7 +39,8 @@ func (hls *HttpLookupSource) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (hls *HttpLookupSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (hls *HttpLookupSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/http/rest_sink.go b/internal/io/http/rest_sink.go
index ba214025..580157ee 100644
--- a/internal/io/http/rest_sink.go
+++ b/internal/io/http/rest_sink.go
@@ -39,7 +39,8 @@ func (r *RestSink) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (r *RestSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (r *RestSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/memory/lookupsource.go b/internal/io/memory/lookupsource.go
index b4ccc0b3..8560812c 100644
--- a/internal/io/memory/lookupsource.go
+++ b/internal/io/memory/lookupsource.go
@@ -43,7 +43,12 @@ func (s *lookupsource) Connect(ctx api.StreamContext, sch api.StatusChangeHandle
 	ctx.GetLogger().Infof("lookup source %s is opened with key %v", s.topic, s.key)
 	var err error
 	s.table, err = store.Reg(s.topic, s.topicRegex, s.key)
-	return err
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
+	}
+	sch(api.ConnectionConnected, "")
+	return nil
 }
 
 func (s *lookupsource) Provision(ctx api.StreamContext, props map[string]any) error {
diff --git a/internal/io/memory/sink.go b/internal/io/memory/sink.go
index ff5347c6..a8776c60 100644
--- a/internal/io/memory/sink.go
+++ b/internal/io/memory/sink.go
@@ -61,9 +61,10 @@ func (s *sink) Provision(_ api.StreamContext, props map[string]any) error {
 	return nil
 }
 
-func (s *sink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *sink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	ctx.GetLogger().Debugf("Opening memory sink: %v", s.topic)
 	pubsub.CreatePub(s.topic)
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/memory/source.go b/internal/io/memory/source.go
index 3db5f6d3..ada560e2 100644
--- a/internal/io/memory/source.go
+++ b/internal/io/memory/source.go
@@ -59,8 +59,8 @@ func (s *source) Provision(ctx api.StreamContext, props map[string]any) error {
 	return nil
 }
 
-func (s *source) Connect(_ api.StreamContext, _ api.StatusChangeHandler) error {
-	// do nothing
+func (s *source) Connect(_ api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index b5de529f..a5d02e50 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -74,7 +74,7 @@ func (conn *Connection) Provision(ctx api.StreamContext, conId string, props map
 	}
 	opts = opts.SetClientID(c.ClientId).SetAutoReconnect(true).SetResumeSubs(true).SetMaxReconnectInterval(connection.DefaultMaxInterval)
 
-	conn.status.Store(api.ConnectionConnecting)
+	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnecting})
 	opts.OnConnect = conn.onConnect
 	opts.OnConnectionLost = conn.onConnectLost
 	opts.OnReconnecting = conn.onReconnecting
@@ -102,17 +102,17 @@ func (conn *Connection) Dial(ctx api.StreamContext) error {
 	return nil
 }
 
-func (conn *Connection) Status() string {
-	return conn.status.Load().(string)
+func (conn *Connection) Status(_ api.StreamContext) modules.ConnectionStatus {
+	return conn.status.Load().(modules.ConnectionStatus)
 }
 
-func (conn *Connection) SetStatusChangeHandler(sch api.StatusChangeHandler) {
+func (conn *Connection) SetStatusChangeHandler(ctx api.StreamContext, sch api.StatusChangeHandler) {
 	conn.scHandler = sch
 }
 
 func (conn *Connection) onConnect(_ pahoMqtt.Client) {
 	conn.connected.Store(true)
-	conn.status.Store(api.ConnectionConnected)
+	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnected})
 	if conn.scHandler != nil {
 		conn.scHandler(api.ConnectionConnected, "")
 	}
@@ -121,7 +121,7 @@ func (conn *Connection) onConnect(_ pahoMqtt.Client) {
 
 func (conn *Connection) onConnectLost(_ pahoMqtt.Client, err error) {
 	conn.connected.Store(false)
-	conn.status.Store(api.ConnectionDisconnected)
+	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionDisconnected, ErrMsg: err.Error()})
 	if conn.scHandler != nil {
 		conn.scHandler(api.ConnectionDisconnected, err.Error())
 	}
@@ -129,7 +129,7 @@ func (conn *Connection) onConnectLost(_ pahoMqtt.Client, err error) {
 }
 
 func (conn *Connection) onReconnecting(_ pahoMqtt.Client, _ *pahoMqtt.ClientOptions) {
-	conn.status.Store(api.ConnectionConnecting)
+	conn.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnecting})
 	if conn.scHandler != nil {
 		conn.scHandler(api.ConnectionConnecting, "")
 	}
@@ -226,3 +226,5 @@ func getTopicFromProps(props map[string]any) (string, error) {
 	}
 	return "", fmt.Errorf("topic or datasource not defined")
 }
+
+var _ modules.StatefulDialer = &Connection{}
diff --git a/internal/io/redis/lookup.go b/internal/io/redis/lookup.go
index e6ffceec..fbc140bb 100644
--- a/internal/io/redis/lookup.go
+++ b/internal/io/redis/lookup.go
@@ -63,7 +63,7 @@ func (s *lookupSource) Provision(ctx api.StreamContext, props map[string]any) er
 	return s.Validate(props)
 }
 
-func (s *lookupSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *lookupSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	logger := ctx.GetLogger()
 	logger.Debug("Opening redis lookup source")
 
@@ -74,7 +74,12 @@ func (s *lookupSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler)
 		DB:       s.db, // use default DB
 	})
 	_, err := s.cli.Ping(ctx).Result()
-	return err
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
+	}
+	sch(api.ConnectionConnected, "")
+	return nil
 }
 
 func (s *lookupSource) Lookup(ctx api.StreamContext, _ []string, keys []string, values []any) ([]map[string]any, error) {
diff --git a/internal/io/redis/redisPub.go b/internal/io/redis/redisPub.go
index bfed2136..99308c94 100644
--- a/internal/io/redis/redisPub.go
+++ b/internal/io/redis/redisPub.go
@@ -74,7 +74,7 @@ func (r *redisPub) Provision(ctx api.StreamContext, props map[string]any) error
 	return r.Validate(props)
 }
 
-func (r *redisPub) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (r *redisPub) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	ctx.GetLogger().Infof("redisSub is opening")
 	r.conn = redis.NewClient(&redis.Options{
 		Addr:     r.conf.Address,
@@ -83,7 +83,12 @@ func (r *redisPub) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) err
 		DB:       r.conf.Db,
 	})
 	_, err := r.conn.Ping(ctx).Result()
-	return err
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
+	}
+	sch(api.ConnectionConnected, "")
+	return nil
 }
 
 func (r *redisPub) Collect(ctx api.StreamContext, item api.RawTuple) error {
diff --git a/internal/io/redis/redisSub.go b/internal/io/redis/redisSub.go
index afb65af0..08a607e6 100644
--- a/internal/io/redis/redisSub.go
+++ b/internal/io/redis/redisSub.go
@@ -71,7 +71,7 @@ func (r *redisSub) Provision(ctx api.StreamContext, props map[string]any) error
 	return r.Validate(props)
 }
 
-func (r *redisSub) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (r *redisSub) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	ctx.GetLogger().Infof("redisSub is opening")
 	r.conn = redis.NewClient(&redis.Options{
 		Addr:     r.conf.Address,
@@ -80,7 +80,12 @@ func (r *redisSub) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) err
 		DB:       r.conf.Db,
 	})
 	_, err := r.conn.Ping(ctx).Result()
-	return err
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
+	}
+	sch(api.ConnectionConnected, "")
+	return nil
 }
 
 func (r *redisSub) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, _ api.ErrorIngest) error {
diff --git a/internal/io/redis/sink.go b/internal/io/redis/sink.go
index d5b5a641..c14524d9 100644
--- a/internal/io/redis/sink.go
+++ b/internal/io/redis/sink.go
@@ -58,7 +58,7 @@ func (r *RedisSink) Provision(_ api.StreamContext, props map[string]any) error {
 	return r.Validate(props)
 }
 
-func (r *RedisSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (r *RedisSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	logger := ctx.GetLogger()
 	logger.Debug("Opening redis sink")
 
@@ -69,7 +69,12 @@ func (r *RedisSink) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) er
 		DB:       r.c.Db, // use default DB
 	})
 	_, err := r.cli.Ping(ctx).Result()
-	return err
+	if err != nil {
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
+	}
+	sch(api.ConnectionConnected, "")
+	return nil
 }
 
 func (r *RedisSink) Validate(props map[string]any) error {
diff --git a/internal/io/simulator/simulator.go b/internal/io/simulator/simulator.go
index fd6d94af..acddcdc7 100644
--- a/internal/io/simulator/simulator.go
+++ b/internal/io/simulator/simulator.go
@@ -46,7 +46,8 @@ func (s *SimulatorSource) Close(ctx api.StreamContext) error {
 	return nil
 }
 
-func (s *SimulatorSource) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (s *SimulatorSource) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/io/websocket/websocket_sink.go b/internal/io/websocket/websocket_sink.go
index 4c9ea81b..21906758 100644
--- a/internal/io/websocket/websocket_sink.go
+++ b/internal/io/websocket/websocket_sink.go
@@ -52,9 +52,10 @@ func (w *WebsocketSink) Close(ctx api.StreamContext) error {
 	return connection.DetachConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint))
 }
 
-func (w *WebsocketSink) Connect(ctx api.StreamContext, sc api.StatusChangeHandler) error {
+func (w *WebsocketSink) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	var err error
-	w.cw, err = connection.FetchConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), "websocket", w.props, sc)
+	// Connection pool will handle status change
+	w.cw, err = connection.FetchConnection(ctx, buildWebsocketEpID(w.cfg.Endpoint), "websocket", w.props, sch)
 	if err != nil {
 		return err
 	}
diff --git a/internal/topo/collector/func.go b/internal/topo/collector/func.go
index 819457bc..c7005dbb 100644
--- a/internal/topo/collector/func.go
+++ b/internal/topo/collector/func.go
@@ -46,12 +46,14 @@ func (c *FuncCollector) Provision(ctx api.StreamContext, configs map[string]any)
 	return nil
 }
 
-func (c *FuncCollector) Connect(ctx api.StreamContext, _ api.StatusChangeHandler) error {
+func (c *FuncCollector) Connect(ctx api.StreamContext, sch api.StatusChangeHandler) error {
 	ctx.GetLogger().Info("Opening func collector")
-
 	if c.f == nil {
-		return errors.New("func collector missing function")
+		err := errors.New("func collector missing function")
+		sch(api.ConnectionDisconnected, err.Error())
+		return err
 	}
+	sch(api.ConnectionConnected, "")
 	return nil
 }
 
diff --git a/internal/topo/lookup/table.go b/internal/topo/lookup/table.go
index bac13cd9..1f8ece9b 100644
--- a/internal/topo/lookup/table.go
+++ b/internal/topo/lookup/table.go
@@ -84,7 +84,9 @@ func CreateInstance(name string, sourceType string, options *ast.Options) error
 	}
 	ctx.GetLogger().Debugf("lookup source %s is configured", sourceType)
 	// TODO lookup table connection status support
-	err = ns.Connect(ctx, nil)
+	err = ns.Connect(ctx, func(status string, message string) {
+		// do nothing
+	})
 	if err != nil {
 		return err
 	}
diff --git a/pkg/nng/sock.go b/pkg/nng/sock.go
index a9f4d681..59ae59ff 100644
--- a/pkg/nng/sock.go
+++ b/pkg/nng/sock.go
@@ -42,10 +42,20 @@ type Sock struct {
 	mangos.Socket
 	url       string
 	id        string
+	scHandler api.StatusChangeHandler
 	connected atomic.Bool
+	status    atomic.Value
 }
 
-func (s *Sock) GetId(ctx api.StreamContext) string {
+func (s *Sock) SetStatusChangeHandler(_ api.StreamContext, handler api.StatusChangeHandler) {
+	s.scHandler = handler
+}
+
+func (s *Sock) Status(_ api.StreamContext) modules.ConnectionStatus {
+	return s.status.Load().(modules.ConnectionStatus)
+}
+
+func (s *Sock) GetId(_ api.StreamContext) string {
 	return s.id
 }
 
@@ -79,13 +89,24 @@ func (s *Sock) Provision(ctx api.StreamContext, conId string, props map[string]a
 		switch ev {
 		case mangos.PipeEventAttached:
 			s.connected.Store(true)
+			s.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnected})
+			if s.scHandler != nil {
+				s.scHandler(api.ConnectionConnected, "")
+			}
 			ctx.GetLogger().Infof("nano connection attached")
 		case mangos.PipeEventAttaching:
-			ctx.GetLogger().Infof("nano connection is attaching")
+			s.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnecting})
+			if s.scHandler != nil {
+				s.scHandler(api.ConnectionConnecting, "")
+			}
+			ctx.GetLogger().Debugf("nano connection is attaching")
 		case mangos.PipeEventDetached:
 			s.connected.Store(false)
+			s.status.Store(modules.ConnectionStatus{Status: api.ConnectionDisconnected})
+			if s.scHandler != nil {
+				s.scHandler(api.ConnectionDisconnected, "")
+			}
 			ctx.GetLogger().Warnf("nano connection detached")
-			// TODO how to let connection send error event?
 		}
 	})
 	return nil
@@ -108,8 +129,11 @@ func (s *Sock) Dial(ctx api.StreamContext) error {
 
 var nngTimeout = 5 * time.Second
 
-func CreateConnection(ctx api.StreamContext) modules.Connection {
-	return &Sock{}
+func CreateConnection(_ api.StreamContext) modules.Connection {
+	s := &Sock{}
+	s.status.Store(modules.ConnectionStatus{Status: api.ConnectionConnecting})
+	s.connected.Store(false)
+	return s
 }
 
 func ValidateConf(props map[string]any) (*SockConf, error) {
@@ -157,4 +181,4 @@ func (s *Sock) Send(ctx api.StreamContext, data []byte) error {
 	return errorx.NewIOErr(`nng connection is not established`)
 }
 
-var _ modules.Connection = &Sock{}
+var _ modules.StatefulDialer = &Sock{}


Test patch:
diff --git a/internal/topo/rule/state_test.go b/internal/topo/rule/state_test.go
index 433e17e0..d8deeeba 100644
--- a/internal/topo/rule/state_test.go
+++ b/internal/topo/rule/state_test.go
@@ -15,6 +15,7 @@
 package rule
 
 import (
+	"regexp"
 	"sync"
 	"testing"
 	"time"
@@ -83,16 +84,22 @@ func TestAPIs(t *testing.T) {
 	}
 	assert.Equal(t, expTopo, topo)
 	sm = st.GetStatusMessage()
-	em := "{\n  \"status\": \"running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 0,\n  \"source_demo_0_connection_last_connected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_project_0_records_in_total\": 0,\n  \"op_2_project_0_records_out_total\": 0,\n  \"op_2_project_0_messages_processed_total\": 0,\n  \"op_2_project_0_process_latency_us\": 0,\n  \"op_2_project_0_buffer_length\": 0,\n  \"op_2_project_0_last_invocation\": 0,\n  \"op_2_project_0_exceptions_total\": 0,\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 0,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
-	assert.Equal(t, em, sm)
+	em := "{\n  \"status\": \"running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 1,\n  \"source_demo_0_connection_last_connected_time\": 1,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_project_0_records_in_total\": 0,\n  \"op_2_project_0_records_out_total\": 0,\n  \"op_2_project_0_messages_processed_total\": 0,\n  \"op_2_project_0_process_latency_us\": 0,\n  \"op_2_project_0_buffer_length\": 0,\n  \"op_2_project_0_last_invocation\": 0,\n  \"op_2_project_0_exceptions_total\": 0,\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 1,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 1,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
+	re := regexp.MustCompile(`connection_last_connected_time":\s*\d+`)
+	rsm := re.ReplaceAllString(sm, `connection_last_connected_time": 1`)
+	assert.Equal(t, em, rsm)
 	mm := st.GetStatusMap()
 	assert.True(t, len(mm) > 0)
 	// Check stop metrics
 	st.Stop()
 	assert.Equal(t, expTopo, st.GetTopoGraph())
-	assert.Equal(t, map[string]any{"lastStartTimestamp": int64(0), "lastStopTimestamp": int64(0), "message": "canceled manually", "nextStartTimestamp": int64(0), "op_2_project_0_buffer_length": int64(0), "op_2_project_0_exceptions_total": int64(0), "op_2_project_0_last_exception": "", "op_2_project_0_last_exception_time": int64(0), "op_2_project_0_last_invocation": int64(0), "op_2_project_0_messages_processed_total": int64(0), "op_2_project_0_process_latency_us": int64(0), "op_2_project_0_records_in_total": int64(0), "op_2_project_0_records_out_total": int64(0), "op_logToMemory_0_0_transform_0_buffer_length": int64(0), "op_logToMemory_0_0_transform_0_exceptions_total": int64(0), "op_logToMemory_0_0_transform_0_last_exception": "", "op_logToMemory_0_0_transform_0_last_exception_time": int64(0), "op_logToMemory_0_0_transform_0_last_invocation": int64(0), "op_logToMemory_0_0_transform_0_messages_processed_total": int64(0), "op_logToMemory_0_0_transform_0_process_latency_us": int64(0), "op_logToMemory_0_0_transform_0_records_in_total": int64(0), "op_logToMemory_0_0_transform_0_records_out_total": int64(0), "op_logToMemory_0_1_encode_0_buffer_length": int64(0), "op_logToMemory_0_1_encode_0_exceptions_total": int64(0), "op_logToMemory_0_1_encode_0_last_exception": "", "op_logToMemory_0_1_encode_0_last_exception_time": int64(0), "op_logToMemory_0_1_encode_0_last_invocation": int64(0), "op_logToMemory_0_1_encode_0_messages_processed_total": int64(0), "op_logToMemory_0_1_encode_0_process_latency_us": int64(0), "op_logToMemory_0_1_encode_0_records_in_total": int64(0), "op_logToMemory_0_1_encode_0_records_out_total": int64(0), "sink_logToMemory_0_0_buffer_length": int64(0), "sink_logToMemory_0_0_exceptions_total": int64(0), "sink_logToMemory_0_0_last_exception": "", "sink_logToMemory_0_0_last_exception_time": int64(0), "sink_logToMemory_0_0_last_invocation": int64(0), "sink_logToMemory_0_0_messages_processed_total": int64(0), "sink_logToMemory_0_0_process_latency_us": int64(0), "sink_logToMemory_0_0_records_in_total": int64(0), "sink_logToMemory_0_0_records_out_total": int64(0), "sink_logToMemory_0_0_connection_last_connected_time": int64(0), "sink_logToMemory_0_0_connection_last_disconnected_message": "", "sink_logToMemory_0_0_connection_last_disconnected_time": int64(0), "sink_logToMemory_0_0_connection_last_try_time": int64(0), "sink_logToMemory_0_0_connection_status": 0, "source_demo_0_buffer_length": int64(0), "source_demo_0_exceptions_total": int64(0), "source_demo_0_last_exception": "", "source_demo_0_last_exception_time": int64(0), "source_demo_0_last_invocation": int64(0), "source_demo_0_messages_processed_total": int64(0), "source_demo_0_process_latency_us": int64(0), "source_demo_0_records_in_total": int64(0), "source_demo_0_records_out_total": int64(0), "source_demo_0_connection_last_connected_time": int64(0), "source_demo_0_connection_last_disconnected_message": "", "source_demo_0_connection_last_disconnected_time": int64(0), "source_demo_0_connection_last_try_time": int64(0), "source_demo_0_connection_status": 0, "status": "stopped"}, st.GetStatusMap())
-	em = "{\n  \"status\": \"stopped\",\n  \"message\": \"canceled manually\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 0,\n  \"source_demo_0_connection_last_connected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_project_0_records_in_total\": 0,\n  \"op_2_project_0_records_out_total\": 0,\n  \"op_2_project_0_messages_processed_total\": 0,\n  \"op_2_project_0_process_latency_us\": 0,\n  \"op_2_project_0_buffer_length\": 0,\n  \"op_2_project_0_last_invocation\": 0,\n  \"op_2_project_0_exceptions_total\": 0,\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 0,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
-	assert.Equal(t, em, st.GetStatusMessage())
+	ssm := st.GetStatusMap()
+	ssm["sink_logToMemory_0_0_connection_last_connected_time"] = int64(1)
+	ssm["source_demo_0_connection_last_connected_time"] = int64(1)
+	assert.Equal(t, map[string]any{"lastStartTimestamp": int64(0), "lastStopTimestamp": int64(0), "message": "canceled manually", "nextStartTimestamp": int64(0), "op_2_project_0_buffer_length": int64(0), "op_2_project_0_exceptions_total": int64(0), "op_2_project_0_last_exception": "", "op_2_project_0_last_exception_time": int64(0), "op_2_project_0_last_invocation": int64(0), "op_2_project_0_messages_processed_total": int64(0), "op_2_project_0_process_latency_us": int64(0), "op_2_project_0_records_in_total": int64(0), "op_2_project_0_records_out_total": int64(0), "op_logToMemory_0_0_transform_0_buffer_length": int64(0), "op_logToMemory_0_0_transform_0_exceptions_total": int64(0), "op_logToMemory_0_0_transform_0_last_exception": "", "op_logToMemory_0_0_transform_0_last_exception_time": int64(0), "op_logToMemory_0_0_transform_0_last_invocation": int64(0), "op_logToMemory_0_0_transform_0_messages_processed_total": int64(0), "op_logToMemory_0_0_transform_0_process_latency_us": int64(0), "op_logToMemory_0_0_transform_0_records_in_total": int64(0), "op_logToMemory_0_0_transform_0_records_out_total": int64(0), "op_logToMemory_0_1_encode_0_buffer_length": int64(0), "op_logToMemory_0_1_encode_0_exceptions_total": int64(0), "op_logToMemory_0_1_encode_0_last_exception": "", "op_logToMemory_0_1_encode_0_last_exception_time": int64(0), "op_logToMemory_0_1_encode_0_last_invocation": int64(0), "op_logToMemory_0_1_encode_0_messages_processed_total": int64(0), "op_logToMemory_0_1_encode_0_process_latency_us": int64(0), "op_logToMemory_0_1_encode_0_records_in_total": int64(0), "op_logToMemory_0_1_encode_0_records_out_total": int64(0), "sink_logToMemory_0_0_buffer_length": int64(0), "sink_logToMemory_0_0_exceptions_total": int64(0), "sink_logToMemory_0_0_last_exception": "", "sink_logToMemory_0_0_last_exception_time": int64(0), "sink_logToMemory_0_0_last_invocation": int64(0), "sink_logToMemory_0_0_messages_processed_total": int64(0), "sink_logToMemory_0_0_process_latency_us": int64(0), "sink_logToMemory_0_0_records_in_total": int64(0), "sink_logToMemory_0_0_records_out_total": int64(0), "sink_logToMemory_0_0_connection_last_connected_time": int64(1), "sink_logToMemory_0_0_connection_last_disconnected_message": "", "sink_logToMemory_0_0_connection_last_disconnected_time": int64(0), "sink_logToMemory_0_0_connection_last_try_time": int64(0), "sink_logToMemory_0_0_connection_status": 1, "source_demo_0_buffer_length": int64(0), "source_demo_0_exceptions_total": int64(0), "source_demo_0_last_exception": "", "source_demo_0_last_exception_time": int64(0), "source_demo_0_last_invocation": int64(0), "source_demo_0_messages_processed_total": int64(0), "source_demo_0_process_latency_us": int64(0), "source_demo_0_records_in_total": int64(0), "source_demo_0_records_out_total": int64(0), "source_demo_0_connection_last_connected_time": int64(1), "source_demo_0_connection_last_disconnected_message": "", "source_demo_0_connection_last_disconnected_time": int64(0), "source_demo_0_connection_last_try_time": int64(0), "source_demo_0_connection_status": 1, "status": "stopped"}, ssm)
+	em = "{\n  \"status\": \"stopped\",\n  \"message\": \"canceled manually\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 1,\n  \"source_demo_0_connection_last_connected_time\": 1,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_project_0_records_in_total\": 0,\n  \"op_2_project_0_records_out_total\": 0,\n  \"op_2_project_0_messages_processed_total\": 0,\n  \"op_2_project_0_process_latency_us\": 0,\n  \"op_2_project_0_buffer_length\": 0,\n  \"op_2_project_0_last_invocation\": 0,\n  \"op_2_project_0_exceptions_total\": 0,\n  \"op_2_project_0_last_exception\": \"\",\n  \"op_2_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 1,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 1,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
+	rsm = re.ReplaceAllString(st.GetStatusMessage(), `connection_last_connected_time": 1`)
+	assert.Equal(t, em, rsm)
 	assert.Equal(t, Stopped, st.currentState)
 	// Update rule
 	st.Rule = def.GetDefaultRule("testAPI", "select abc from demo where a > 3")
@@ -128,8 +135,9 @@ func TestAPIs(t *testing.T) {
 	}
 	assert.Equal(t, expTopo, topo)
 	sm = st.GetStatusMessage()
-	em = "{\n  \"status\": \"running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 0,\n  \"source_demo_0_connection_last_connected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_filter_0_records_in_total\": 0,\n  \"op_2_filter_0_records_out_total\": 0,\n  \"op_2_filter_0_messages_processed_total\": 0,\n  \"op_2_filter_0_process_latency_us\": 0,\n  \"op_2_filter_0_buffer_length\": 0,\n  \"op_2_filter_0_last_invocation\": 0,\n  \"op_2_filter_0_exceptions_total\": 0,\n  \"op_2_filter_0_last_exception\": \"\",\n  \"op_2_filter_0_last_exception_time\": 0,\n  \"op_3_project_0_records_in_total\": 0,\n  \"op_3_project_0_records_out_total\": 0,\n  \"op_3_project_0_messages_processed_total\": 0,\n  \"op_3_project_0_process_latency_us\": 0,\n  \"op_3_project_0_buffer_length\": 0,\n  \"op_3_project_0_last_invocation\": 0,\n  \"op_3_project_0_exceptions_total\": 0,\n  \"op_3_project_0_last_exception\": \"\",\n  \"op_3_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 0,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
-	assert.Equal(t, em, sm)
+	rsm = re.ReplaceAllString(sm, `connection_last_connected_time": 1`)
+	em = "{\n  \"status\": \"running\",\n  \"message\": \"\",\n  \"lastStartTimestamp\": 0,\n  \"lastStopTimestamp\": 0,\n  \"nextStartTimestamp\": 0,\n  \"source_demo_0_records_in_total\": 0,\n  \"source_demo_0_records_out_total\": 0,\n  \"source_demo_0_messages_processed_total\": 0,\n  \"source_demo_0_process_latency_us\": 0,\n  \"source_demo_0_buffer_length\": 0,\n  \"source_demo_0_last_invocation\": 0,\n  \"source_demo_0_exceptions_total\": 0,\n  \"source_demo_0_last_exception\": \"\",\n  \"source_demo_0_last_exception_time\": 0,\n  \"source_demo_0_connection_status\": 1,\n  \"source_demo_0_connection_last_connected_time\": 1,\n  \"source_demo_0_connection_last_disconnected_time\": 0,\n  \"source_demo_0_connection_last_disconnected_message\": \"\",\n  \"source_demo_0_connection_last_try_time\": 0,\n  \"op_2_filter_0_records_in_total\": 0,\n  \"op_2_filter_0_records_out_total\": 0,\n  \"op_2_filter_0_messages_processed_total\": 0,\n  \"op_2_filter_0_process_latency_us\": 0,\n  \"op_2_filter_0_buffer_length\": 0,\n  \"op_2_filter_0_last_invocation\": 0,\n  \"op_2_filter_0_exceptions_total\": 0,\n  \"op_2_filter_0_last_exception\": \"\",\n  \"op_2_filter_0_last_exception_time\": 0,\n  \"op_3_project_0_records_in_total\": 0,\n  \"op_3_project_0_records_out_total\": 0,\n  \"op_3_project_0_messages_processed_total\": 0,\n  \"op_3_project_0_process_latency_us\": 0,\n  \"op_3_project_0_buffer_length\": 0,\n  \"op_3_project_0_last_invocation\": 0,\n  \"op_3_project_0_exceptions_total\": 0,\n  \"op_3_project_0_last_exception\": \"\",\n  \"op_3_project_0_last_exception_time\": 0,\n  \"op_logToMemory_0_0_transform_0_records_in_total\": 0,\n  \"op_logToMemory_0_0_transform_0_records_out_total\": 0,\n  \"op_logToMemory_0_0_transform_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_0_transform_0_process_latency_us\": 0,\n  \"op_logToMemory_0_0_transform_0_buffer_length\": 0,\n  \"op_logToMemory_0_0_transform_0_last_invocation\": 0,\n  \"op_logToMemory_0_0_transform_0_exceptions_total\": 0,\n  \"op_logToMemory_0_0_transform_0_last_exception\": \"\",\n  \"op_logToMemory_0_0_transform_0_last_exception_time\": 0,\n  \"op_logToMemory_0_1_encode_0_records_in_total\": 0,\n  \"op_logToMemory_0_1_encode_0_records_out_total\": 0,\n  \"op_logToMemory_0_1_encode_0_messages_processed_total\": 0,\n  \"op_logToMemory_0_1_encode_0_process_latency_us\": 0,\n  \"op_logToMemory_0_1_encode_0_buffer_length\": 0,\n  \"op_logToMemory_0_1_encode_0_last_invocation\": 0,\n  \"op_logToMemory_0_1_encode_0_exceptions_total\": 0,\n  \"op_logToMemory_0_1_encode_0_last_exception\": \"\",\n  \"op_logToMemory_0_1_encode_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_records_in_total\": 0,\n  \"sink_logToMemory_0_0_records_out_total\": 0,\n  \"sink_logToMemory_0_0_messages_processed_total\": 0,\n  \"sink_logToMemory_0_0_process_latency_us\": 0,\n  \"sink_logToMemory_0_0_buffer_length\": 0,\n  \"sink_logToMemory_0_0_last_invocation\": 0,\n  \"sink_logToMemory_0_0_exceptions_total\": 0,\n  \"sink_logToMemory_0_0_last_exception\": \"\",\n  \"sink_logToMemory_0_0_last_exception_time\": 0,\n  \"sink_logToMemory_0_0_connection_status\": 1,\n  \"sink_logToMemory_0_0_connection_last_connected_time\": 1,\n  \"sink_logToMemory_0_0_connection_last_disconnected_time\": 0,\n  \"sink_logToMemory_0_0_connection_last_disconnected_message\": \"\",\n  \"sink_logToMemory_0_0_connection_last_try_time\": 0\n}"
+	assert.Equal(t, em, rsm)
 	e = st.Delete()
 	assert.NoError(t, e)
 }
diff --git a/pkg/nng/sock_test.go b/pkg/nng/sock_test.go
index 4d4e132f..456443f7 100644
--- a/pkg/nng/sock_test.go
+++ b/pkg/nng/sock_test.go
@@ -16,8 +16,16 @@ package nng
 
 import (
 	"testing"
+	"time"
 
+	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.nanomsg.org/mangos/v3/protocol/pair"
+	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 func TestValidate(t *testing.T) {
@@ -62,3 +70,73 @@ func TestValidate(t *testing.T) {
 		})
 	}
 }
+
+func TestConStatus(t *testing.T) {
+	var statusHistory []modules.ConnectionStatus
+	scRecorder := func(status string, message string) {
+		statusHistory = append(statusHistory, modules.ConnectionStatus{Status: status, ErrMsg: message})
+	}
+	ctx := mockContext.NewMockContext("testConStatus", "test")
+	c := CreateConnection(ctx).(*Sock)
+	err := c.Provision(ctx, "testconstatus", map[string]any{
+		"url": "ipc:///tmp/testconstatus.ipc",
+	})
+	require.NoError(t, err)
+	c.SetStatusChangeHandler(ctx, scRecorder)
+	err = c.Dial(ctx)
+	require.NoError(t, err)
+	defer c.Close(ctx)
+	st := c.Status(ctx)
+	require.Equal(t, modules.ConnectionStatus{Status: api.ConnectionConnecting}, st)
+	// Connect
+	sock, err := pair.NewSocket()
+	require.NoError(t, err)
+	err = sock.Listen("ipc:///tmp/testconstatus.ipc")
+	require.NoError(t, err)
+	retry := 10
+	for i := 0; i < retry; i++ {
+		time.Sleep(100 * time.Millisecond)
+		if c.connected.Load() {
+			break
+		}
+	}
+	if c.connected.Load() {
+		st = c.Status(ctx)
+		assert.Equal(t, modules.ConnectionStatus{Status: api.ConnectionConnected}, st)
+	} else {
+		require.FailNow(t, "failed to connect")
+	}
+	// Disconnect
+	err = sock.Close()
+	require.NoError(t, err)
+	for i := 0; i < retry; i++ {
+		time.Sleep(100 * time.Millisecond)
+		if !c.connected.Load() {
+			break
+		}
+	}
+	if !c.connected.Load() {
+		st = c.Status(ctx)
+		assert.Equal(t, modules.ConnectionStatus{Status: api.ConnectionDisconnected}, st)
+	} else {
+		require.FailNow(t, "failed to connect")
+	}
+	// ReConnect
+	sock, err = pair.NewSocket()
+	require.NoError(t, err)
+	err = sock.Listen("ipc:///tmp/testconstatus.ipc")
+	require.NoError(t, err)
+	retry = 10
+	for i := 0; i < retry; i++ {
+		time.Sleep(100 * time.Millisecond)
+		if c.connected.Load() {
+			break
+		}
+	}
+	if c.connected.Load() {
+		st = c.Status(ctx)
+		assert.Equal(t, modules.ConnectionStatus{Status: api.ConnectionConnected}, st)
+	} else {
+		require.FailNow(t, "failed to connect")
+	}
+}