support status

Signed-off-by: yisaer <disxiaofei@163.com>

internal/binder/io/binder.go

@@ -20,7 +20,6 @@ import (
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/binder"
-	mqttClient "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 )
 
 var ( // init once and read only
@@ -100,7 +99,3 @@ func LookupSource(name string) (api.LookupSource, error) {
 	}
 	return nil, errs
 }
-
-func RegisterConnection() {
-	mqttClient.Register()
-}

internal/io/connection/connection_mock.go

@@ -15,6 +15,9 @@
 package connection
 
 import (
+	"errors"
+	"strings"
+
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
@@ -23,6 +26,7 @@ import (
 func InitMockTest() {
 	conf.IsTesting = true
 	modules.ConnectionRegister["mock"] = CreateMockConnection
+	modules.ConnectionRegister[strings.ToLower("mockErr")] = CreateMockErrConnection
 }
 
 type mockConnection struct {
@@ -61,3 +65,34 @@ func CreateMockConnection(ctx api.StreamContext, id string, props map[string]any
 	m := &mockConnection{id: id, ref: 0}
 	return m, nil
 }
+
+type mockErrConnection struct {
+}
+
+func (m mockErrConnection) Ping(ctx api.StreamContext) error {
+	return errors.New("mockErr")
+}
+
+func (m mockErrConnection) Close(ctx api.StreamContext) {
+	return
+}
+
+func (m mockErrConnection) Attach(ctx api.StreamContext) {
+	return
+}
+
+func (m mockErrConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	return
+}
+
+func (m mockErrConnection) DetachPub(ctx api.StreamContext, props map[string]any) {
+	return
+}
+
+func (m mockErrConnection) Ref(ctx api.StreamContext) int {
+	return 0
+}
+
+func CreateMockErrConnection(ctx api.StreamContext, id string, props map[string]any) (modules.Connection, error) {
+	return nil, errors.New("mockErr")
+}

internal/io/connection/connection_pool.go

@@ -209,6 +209,10 @@ func DropNameConnection(ctx api.StreamContext, selId string) error {
 	defer globalConnectionManager.Unlock()
 	meta, ok := globalConnectionManager.connectionPool[selId]
 	if !ok {
+		_, ok := globalConnectionManager.failConnection[selId]
+		if ok {
+			delete(globalConnectionManager.failConnection, selId)
+		}
 		return nil
 	}
 	conn := meta.conn
@@ -285,10 +289,6 @@ type ConnectionMeta struct {
 	conn  modules.Connection `json:"-"`
 }
 
-func init() {
-	modules.ConnectionRegister["mock"] = CreateMockConnection
-}
-
 func NewExponentialBackOff() *backoff.ExponentialBackOff {
 	return backoff.NewExponentialBackOff(
 		backoff.WithInitialInterval(DefaultInitialInterval),

internal/io/connection/connection_pool_test.go

@@ -113,3 +113,27 @@ func TestConnectionErr(t *testing.T) {
 	require.Error(t, err)
 	failpoint.Disable("github.com/lf-edge/ekuiper/v2/internal/io/connection/dropConnectionStoreErr")
 }
+
+func TestConnectionStatus(t *testing.T) {
+	dataDir, err := conf.GetDataLoc()
+	require.NoError(t, err)
+	require.NoError(t, store.SetupDefault(dataDir))
+	require.NoError(t, InitConnectionManager4Test())
+
+	conf.WriteCfgIntoKVStorage("connections", "mockErr", "a1", map[string]interface{}{})
+	conf.WriteCfgIntoKVStorage("connections", "mock", "a2", map[string]interface{}{})
+	require.NoError(t, ReloadConnection())
+	ctx := context.Background()
+	allStatus := GetAllConnectionStatus(ctx)
+	s, ok := allStatus["a1"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionFail,
+		ErrMsg: "mockErr",
+	}, s)
+	s, ok = allStatus["a2"]
+	require.True(t, ok)
+	require.Equal(t, ConnectionStatus{
+		Status: ConnectionRunning,
+	}, s)
+}

internal/server/connection.go

@@ -31,6 +31,16 @@ type ConnectionRequest struct {
 	Props map[string]interface{} `json:"props"`
 }
 
+func connectionsStatusHandler(w http.ResponseWriter, r *http.Request) {
+	defer r.Body.Close()
+	switch r.Method {
+	case http.MethodGet:
+		allStatus := connection.GetAllConnectionStatus(context.Background())
+		w.WriteHeader(http.StatusOK)
+		jsonResponse(allStatus, w, logger)
+	}
+}
+
 func connectionsHandler(w http.ResponseWriter, r *http.Request) {
 	defer r.Body.Close()
 	switch r.Method {

internal/server/rest.go

@@ -177,6 +177,7 @@ func createRestServer(ip string, port int, needToken bool) *http.Server {
 	r.HandleFunc("/data/import", configurationImportHandler).Methods(http.MethodPost)
 	r.HandleFunc("/data/import/status", configurationStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connections", connectionsHandler).Methods(http.MethodGet, http.MethodPost)
+	r.HandleFunc("/connections/status", connectionsStatusHandler).Methods(http.MethodGet)
 	r.HandleFunc("/connection/{id}", connectionHandler).Methods(http.MethodGet, http.MethodDelete)
 	// r.HandleFunc("/connection/websocket", connectionHandler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete)
 	// Register extended routes