From e8b8fa2c83cb6b11c6b0898bf9085c8c09810bc7 Mon Sep 17 00:00:00 2001 From: Jiyong Huang Date: Sat, 11 May 2024 16:58:44 +0800 Subject: [PATCH] fix: detach for non shared connection Signed-off-by: Jiyong Huang --- internal/io/mqtt/connection_pool.go | 39 +++++++++++------------------ internal/io/mqtt/connection_test.go | 12 ++++----- internal/io/mqtt/sink.go | 13 +++++----- internal/io/mqtt/sink_test.go | 12 +++++---- internal/io/mqtt/source.go | 6 ++--- 5 files changed, 36 insertions(+), 46 deletions(-) diff --git a/internal/io/mqtt/connection_pool.go b/internal/io/mqtt/connection_pool.go index cc944a9c8a..e63e12587c 100644 --- a/internal/io/mqtt/connection_pool.go +++ b/internal/io/mqtt/connection_pool.go @@ -15,7 +15,6 @@ package mqtt import ( - "fmt" "sync" "github.com/lf-edge/ekuiper/contract/v2/api" @@ -26,45 +25,35 @@ var ( lock sync.RWMutex ) -func GetConnection(ctx api.StreamContext, props map[string]any) (*Connection, error) { - var clientId string - if sid, ok := props["connectionSelector"]; ok { - if s, ok := sid.(string); ok { - clientId = s - } else { - return nil, fmt.Errorf("connectionSelector value: %v is not string", sid) - } - } - if clientId == "" { +func GetConnection(ctx api.StreamContext, selId string, props map[string]any) (*Connection, error) { + if selId == "" { return CreateClient(ctx, "", props) } lock.Lock() defer lock.Unlock() - if conn, ok := connectionPool[clientId]; ok { + if conn, ok := connectionPool[selId]; ok { conn.attach() return conn, nil } else { - cli, err := CreateClient(ctx, clientId, props) + cli, err := CreateClient(ctx, selId, props) if err != nil { return nil, err } - connectionPool[clientId] = cli + connectionPool[selId] = cli return cli, nil } } -func DetachConnection(clientId string, subscribedTopic string) { +func DetachConnection(conn *Connection, selId string, subscribedTopic string) { + var closed bool + if subscribedTopic != "" { + closed = conn.detachSub(subscribedTopic) + } else { + closed = conn.detachPub() + } lock.Lock() defer lock.Unlock() - if conn, ok := connectionPool[clientId]; ok { - var closed bool - if subscribedTopic != "" { - closed = conn.detachSub(subscribedTopic) - } else { - closed = conn.detachPub() - } - if closed { - delete(connectionPool, clientId) - } + if _, ok := connectionPool[selId]; closed && ok { + delete(connectionPool, selId) } } diff --git a/internal/io/mqtt/connection_test.go b/internal/io/mqtt/connection_test.go index 0b6e22aa93..42704f3080 100644 --- a/internal/io/mqtt/connection_test.go +++ b/internal/io/mqtt/connection_test.go @@ -33,17 +33,17 @@ func TestConnectionLC(t *testing.T) { "server": "abc", } ctx := mockContext.NewMockContext("test", "op") - connShared1, err := GetConnection(ctx, propsShared) + connShared1, err := GetConnection(ctx, "mqtt.localConnection", propsShared) assert.NoError(t, err) assert.Equal(t, 1, len(connectionPool)) - _, err = GetConnection(ctx, propsNormal) + _, err = GetConnection(ctx, "", propsNormal) assert.NoError(t, err) assert.Equal(t, 1, len(connectionPool)) - connShared2, err := GetConnection(ctx, propsShared) + connShared2, err := GetConnection(ctx, "mqtt.localConnection", propsShared) assert.NoError(t, err) assert.Equal(t, 1, len(connectionPool)) assert.Equal(t, connShared1, connShared2) - _, err = GetConnection(ctx, propsInvalid) + _, err = GetConnection(ctx, "", propsInvalid) assert.Error(t, err) assert.Equal(t, err.Error(), "found error when connecting for abc: network Error : dial tcp: address abc: missing port in address") assert.Equal(t, 1, len(connectionPool)) @@ -56,9 +56,9 @@ func TestConnectionLC(t *testing.T) { // Test subscribe in the connector test. - DetachConnection("mqtt.localConnection", "") + DetachConnection(connShared1, "mqtt.localConnection", "") assert.Equal(t, 1, len(connectionPool)) - DetachConnection("mqtt.localConnection", "") + DetachConnection(connShared2, "mqtt.localConnection", "") assert.Equal(t, 0, len(connectionPool)) } diff --git a/internal/io/mqtt/sink.go b/internal/io/mqtt/sink.go index 634f094679..55aba5965a 100644 --- a/internal/io/mqtt/sink.go +++ b/internal/io/mqtt/sink.go @@ -25,10 +25,10 @@ import ( // AdConf is the advanced configuration for the mqtt sink type AdConf struct { - Tpc string `json:"topic"` - Qos byte `json:"qos"` - Retained bool `json:"retained"` - + Tpc string `json:"topic"` + Qos byte `json:"qos"` + Retained bool `json:"retained"` + SelId string `json:"connectionSelector"` ResendTopic string `json:"resendDestination"` } @@ -67,7 +67,7 @@ func (ms *Sink) Provision(_ api.StreamContext, ps map[string]any) error { func (ms *Sink) Connect(ctx api.StreamContext) error { ctx.GetLogger().Infof("Connecting to mqtt server") - cli, err := GetConnection(ctx, ms.config) + cli, err := GetConnection(ctx, ms.adconf.SelId, ms.config) ms.cli = cli return err } @@ -100,8 +100,7 @@ func (ms *Sink) Collect(ctx api.StreamContext, item api.RawTuple) error { func (ms *Sink) Close(ctx api.StreamContext) error { ctx.GetLogger().Info("Closing mqtt sink connector") if ms.cli != nil { - DetachConnection(ms.cli.GetClientId(), "") - ms.cli = nil + DetachConnection(ms.cli, ms.adconf.SelId, "") } return nil } diff --git a/internal/io/mqtt/sink_test.go b/internal/io/mqtt/sink_test.go index 83afccd077..da9a579c26 100644 --- a/internal/io/mqtt/sink_test.go +++ b/internal/io/mqtt/sink_test.go @@ -87,17 +87,19 @@ func TestSinkConfigure(t *testing.T) { { name: "Valid configuration with QoS 0 and no compression", input: map[string]interface{}{ - "topic": "testTopic3", - "qos": 0, - "retained": false, - "compression": "", - "server": "123", + "topic": "testTopic3", + "qos": 0, + "retained": false, + "compression": "", + "server": "123", + "connectionSelector": "mqtt.local", }, expectedAdConf: &AdConf{ Tpc: "testTopic3", Qos: 0, Retained: false, ResendTopic: "testTopic3", + SelId: "mqtt.local", }, }, { diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go index 7c364f3b61..8c132bef8a 100644 --- a/internal/io/mqtt/source.go +++ b/internal/io/mqtt/source.go @@ -38,6 +38,7 @@ type SourceConnector struct { type Conf struct { Topic string `json:"datasource"` Qos int `json:"qos"` + SelId string `json:"connectionSelector"` } func (ms *SourceConnector) Provision(ctx api.StreamContext, props map[string]any) error { @@ -70,7 +71,7 @@ func (ms *SourceConnector) Ping(props map[string]interface{}) error { func (ms *SourceConnector) Connect(ctx api.StreamContext) error { ctx.GetLogger().Infof("Connecting to mqtt server") - cli, err := GetConnection(ctx, ms.props) + cli, err := GetConnection(ctx, ms.cfg.SelId, ms.props) ms.cli = cli return err } @@ -104,8 +105,7 @@ func (ms *SourceConnector) onMessage(ctx api.StreamContext, msg pahoMqtt.Message func (ms *SourceConnector) Close(ctx api.StreamContext) error { ctx.GetLogger().Infof("Closing mqtt source connector to topic %s.", ms.tpc) if ms.cli != nil { - DetachConnection(ms.cli.GetClientId(), ms.tpc) - ms.cli = nil + DetachConnection(ms.cli, ms.cfg.SelId, ms.tpc) } return nil }