Repository: lf-edge/ekuiper
Pull request #2800 (2024-04-24 07:51:52)

feat(api): add error ingest for source subscribe


Patch:
diff --git a/contract/api/source.go b/contract/api/source.go
index 6746e1c4..f5d064d4 100644
--- a/contract/api/source.go
+++ b/contract/api/source.go
@@ -26,11 +26,12 @@ type Source interface {
 	Connector
 }
 
+type ErrorIngest func(ctx StreamContext, err error)
 type BytesIngest func(ctx StreamContext, payload []byte, meta map[string]any, ts time.Time)
 
 type BytesSource interface {
 	Source
-	Subscribe(ctx StreamContext, ingest BytesIngest) error
+	Subscribe(ctx StreamContext, ingest BytesIngest, ingestError ErrorIngest) error
 }
 
 // TupleIngest reads in a structural data or its list.
@@ -39,7 +40,7 @@ type TupleIngest func(ctx StreamContext, data any, meta map[string]any, ts time.
 
 type TupleSource interface {
 	Source
-	Subscribe(ctx StreamContext, ingest TupleIngest) error
+	Subscribe(ctx StreamContext, ingest TupleIngest, ingestError ErrorIngest) error
 }
 
 type EOFIngest func(ctx StreamContext)
diff --git a/internal/io/memory/source.go b/internal/io/memory/source.go
index a138f8de..37b81418 100644
--- a/internal/io/memory/source.go
+++ b/internal/io/memory/source.go
@@ -63,7 +63,7 @@ func (s *source) Connect(_ api.StreamContext) error {
 }
 
 // Subscribe For memory source, it can receive a source tuple directly. So just pass it through
-func (s *source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest) error {
+func (s *source) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, _ api.ErrorIngest) error {
 	ch := pubsub.CreateSub(s.c.Topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.c.BufferLength)
 	ctx.GetLogger().Infof("Subscribe to topic %s", s.c.Topic)
 	go func() {
diff --git a/internal/io/mqtt/source.go b/internal/io/mqtt/source.go
index e834d6e0..5a53f1b9 100644
--- a/internal/io/mqtt/source.go
+++ b/internal/io/mqtt/source.go
@@ -83,15 +83,14 @@ func (ms *SourceConnector) Connect(ctx api.StreamContext) error {
 
 // Subscribe is a one time only operation for source. It connects to the mqtt broker and subscribe to the topic
 // Run open before subscribe
-func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest) error {
+func (ms *SourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
 	return ms.cli.Subscribe(ms.tpc, &SubscriptionInfo{
 		Qos: byte(ms.cfg.Qos),
 		Handler: func(client pahoMqtt.Client, message pahoMqtt.Message) {
 			ms.onMessage(ctx, message, ingest)
 		},
-		// TODO signal ingest
 		ErrHandler: func(err error) {
-			ms.onError(ctx, err)
+			ingestError(ctx, err)
 		},
 	})
 }
@@ -108,10 +107,6 @@ func (ms *SourceConnector) onMessage(ctx api.StreamContext, msg pahoMqtt.Message
 	}, rcvTime)
 }
 
-func (ms *SourceConnector) onError(ctx api.StreamContext, err error) {
-	ctx.GetLogger().Error(err)
-}
-
 func (ms *SourceConnector) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Infof("Closing mqtt source connector to topic %s.", ms.tpc)
 	if ms.cli != nil {
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index c3a6f497..d69c3438 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -115,6 +115,12 @@ func (m *SourceNode) ingestTuple(t api.Tuple, ts time.Time) {
 	m.statManager.IncTotalRecordsOut()
 }
 
+func (m *SourceNode) ingestError(ctx api.StreamContext, err error) {
+	ctx.GetLogger().Error(err)
+	m.Broadcast(err)
+	m.statManager.IncTotalExceptions(err.Error())
+}
+
 func (m *SourceNode) ingestEof(ctx api.StreamContext) {
 	ctx.GetLogger().Infof("send out EOF")
 	m.Broadcast(xsql.EOFTuple(0))
@@ -130,9 +136,9 @@ func (m *SourceNode) Run(ctx api.StreamContext, ctrlCh chan<- error) {
 		}
 		switch ss := m.s.(type) {
 		case api.BytesSource:
-			err = ss.Subscribe(ctx, m.ingestBytes)
+			err = ss.Subscribe(ctx, m.ingestBytes, m.ingestError)
 		case api.TupleSource:
-			err = ss.Subscribe(ctx, m.ingestAnyTuple)
+			err = ss.Subscribe(ctx, m.ingestAnyTuple, m.ingestError)
 		}
 		if err != nil {
 			return err


Test patch:
diff --git a/internal/io/memory/memory_test.go b/internal/io/memory/memory_test.go
index 54c1143f..9d5f72a2 100644
--- a/internal/io/memory/memory_test.go
+++ b/internal/io/memory/memory_test.go
@@ -75,7 +75,7 @@ func TestSharedInmemoryNode(t *testing.T) {
 		expected := []api.Tuple{model.NewDefaultSourceTuple(rawTuple.Message(), xsql.Message{"topic": id}, timex.GetNow())}
 		assert.Equal(t, expected, res)
 		cancel()
-	})
+	}, nil)
 	assert.NoError(t, err)
 	<-ctx.Done()
 }
@@ -213,7 +213,7 @@ func TestMultipleTopics(t *testing.T) {
 				cancel()
 				wg.Done()
 			}
-		})
+		}, nil)
 		assert.NoError(t, err)
 	}
 
diff --git a/internal/topo/node/source_node_test.go b/internal/topo/node/source_node_test.go
index a034624d..48402e9a 100644
--- a/internal/topo/node/source_node_test.go
+++ b/internal/topo/node/source_node_test.go
@@ -190,7 +190,7 @@ func (m *MockSourceConnector) Close(ctx api.StreamContext) error {
 	}
 }
 
-func (m *MockSourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest) error {
+func (m *MockSourceConnector) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
 	if m.subscribed.Load() {
 		return fmt.Errorf("already subscribed")
 	}
diff --git a/internal/topo/topotest/mocknode/mock_source.go b/internal/topo/topotest/mocknode/mock_source.go
index 22d092e6..73210ef2 100644
--- a/internal/topo/topotest/mocknode/mock_source.go
+++ b/internal/topo/topotest/mocknode/mock_source.go
@@ -51,7 +51,7 @@ func (m *MockSource) Connect(_ api.StreamContext) error {
 	return nil
 }
 
-func (m *MockSource) Subscribe(ctx api.StreamContext, ingest api.TupleIngest) error {
+func (m *MockSource) Subscribe(ctx api.StreamContext, ingest api.TupleIngest, ingestError api.ErrorIngest) error {
 	log := ctx.GetLogger()
 	mockClock := timex.Clock
 	log.Infof("%d: mock source %s starts", timex.GetNowInMilli(), ctx.GetOpId())
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index 22c33839..d137d5fb 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -62,6 +62,8 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 				if limit <= 0 {
 					wg.Done()
 				}
+			}, func(ctx api.StreamContext, err error) {
+				panic(err)
 			})
 		case api.TupleSource:
 			panic("added later")