refactor(*): Move infra to pkg and add close for sourcepool

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

cmd/kuiper/main.go

@@ -18,8 +18,8 @@ import (
 	"bufio"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/pkg/model"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"github.com/urfave/cli"
 	"io/ioutil"
 	"net/rpc"

internal/plugin/portable/runtime/plugin_ins_manager.go

@@ -18,8 +18,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"os"
 	"os/exec"
 	"sync"

internal/plugin/portable/runtime/source.go

@@ -17,9 +17,9 @@ package runtime
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"go.nanomsg.org/mangos/v3"
 )
 

internal/processor/rule_query.go

@@ -18,10 +18,10 @@
 package processor
 
 import (
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
 func (p *RuleProcessor) ExecQuery(ruleid, sql string) (*topo.Topo, error) {

internal/server/rpc.go

@@ -23,9 +23,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/pkg/model"
 	"github.com/lf-edge/ekuiper/internal/topo/sink"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"net/http"
 	"net/rpc"
 	"strings"

internal/server/rule_manager.go

@@ -20,11 +20,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/topo"
 	"github.com/lf-edge/ekuiper/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sort"
 	"sync"
 )

internal/service/executors.go

@@ -21,10 +21,10 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/jhump/protoreflect/dynamic"
 	"github.com/jhump/protoreflect/dynamic/grpcdynamic"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/pkg/httpx"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"github.com/ugorji/go/codec"
 	"google.golang.org/grpc"
 	"io/ioutil"

internal/topo/checkpoint/barrier_handler.go

@@ -15,8 +15,8 @@
 package checkpoint
 
 import (
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
 type BarrierHandler interface {

internal/topo/checkpoint/coordinator.go

@@ -17,9 +17,9 @@ package checkpoint
 import (
 	"github.com/benbjohnson/clock"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
 )
 

internal/topo/checkpoint/responder.go

@@ -16,7 +16,7 @@ package checkpoint
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/infra"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
 type Responder interface {

internal/topo/node/join_align_node.go

@@ -16,9 +16,9 @@ package node
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
 // JoinAlignNode will block the stream and buffer all the table tuples. Once buffered, it will combine the later input with the buffer

internal/topo/node/operations.go

@@ -16,9 +16,9 @@ package node
 
 import (
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
 )
 

internal/topo/node/sink_cache.go

@@ -18,10 +18,10 @@ import (
 	"encoding/gob"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"github.com/lf-edge/ekuiper/pkg/kv"
 	"path"
 	"sort"

internal/topo/node/sink_node.go

@@ -18,12 +18,12 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/cast"
 	"github.com/lf-edge/ekuiper/pkg/errorx"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"strings"
 	"sync"
 	"time"

internal/topo/node/source_node.go

@@ -16,11 +16,11 @@ package node
 
 import (
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
 )
 

internal/topo/node/source_pool.go

@@ -19,10 +19,10 @@ import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/binder/io"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
 )
 
@@ -75,7 +75,9 @@ func getSourceInstance(node *SourceNode, index int) (*sourceInstance, error) {
 			}
 			go func() {
 				err := infra.SafeRun(func() error {
-					si.source.Open(node.ctx.WithInstance(index), si.dataCh.In, si.errorCh)
+					nctx := node.ctx.WithInstance(index)
+					defer si.source.Close(nctx)
+					si.source.Open(nctx, si.dataCh.In, si.errorCh)
 					return nil
 				})
 				if err != nil {
@@ -151,7 +153,9 @@ func (p *sourcePool) addInstance(k string, node *SourceNode, source api.Source,
 		p.registry[k] = newS
 		go func() {
 			err := infra.SafeRun(func() error {
-				si.source.Open(node.ctx.WithInstance(index), si.dataCh.In, si.errorCh)
+				nctx := node.ctx.WithInstance(index)
+				defer si.source.Close(nctx)
+				si.source.Open(nctx, si.dataCh.In, si.errorCh)
 				return nil
 			})
 			if err != nil {

internal/topo/node/watermark.go

@@ -17,10 +17,10 @@ package node
 import (
 	"context"
 	"fmt"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"math"
 	"sort"
 )

internal/topo/node/window_op.go

@@ -19,11 +19,11 @@ import (
 	"fmt"
 	"github.com/benbjohnson/clock"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/xsql"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/ast"
 	"github.com/lf-edge/ekuiper/pkg/cast"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"math"
 	"time"
 )

internal/topo/topo.go

@@ -18,12 +18,12 @@ import (
 	"context"
 	"fmt"
 	"github.com/lf-edge/ekuiper/internal/conf"
-	"github.com/lf-edge/ekuiper/internal/infra"
 	"github.com/lf-edge/ekuiper/internal/topo/checkpoint"
 	kctx "github.com/lf-edge/ekuiper/internal/topo/context"
 	"github.com/lf-edge/ekuiper/internal/topo/node"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/infra"
 	"strconv"
 	"sync"
 )