Repository: lf-edge/ekuiper
Pull request #2976 (2024-07-02 10:38:41)

refactor(contract): function exec ctx arg first

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?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 `61.53846%` with `5 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@ed7bd5b`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?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 |
|---|---|---|
| [internal/binder/function/funcs\_trans.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&filepath=internal%2Fbinder%2Ffunction%2Ffuncs_trans.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvYmluZGVyL2Z1bmN0aW9uL2Z1bmNzX3RyYW5zLmdv) | 50.00% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/binder/function/static\_executor.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&filepath=internal%2Fbinder%2Ffunction%2Fstatic_executor.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvYmluZGVyL2Z1bmN0aW9uL3N0YXRpY19leGVjdXRvci5nbw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/binder/mock/mock\_factory.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&filepath=internal%2Fbinder%2Fmock%2Fmock_factory.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvYmluZGVyL21vY2svbW9ja19mYWN0b3J5Lmdv) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/plugin/portable/runtime/function.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&filepath=internal%2Fplugin%2Fportable%2Fruntime%2Ffunction.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvcGx1Z2luL3BvcnRhYmxlL3J1bnRpbWUvZnVuY3Rpb24uZ28=) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?src=pr&el=tree&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-v2    #2976   +/-   ##
============================================
  Coverage             ?   68.77%           
============================================
  Files                ?      329           
  Lines                ?    31676           
  Branches             ?        0           
============================================
  Hits                 ?    21785           
  Misses               ?     7960           
  Partials             ?     1931           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2976?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/Makefile b/Makefile
index 14ccf5c0..c8b7ce7f 100644
--- a/Makefile
+++ b/Makefile
@@ -42,8 +42,8 @@ build_prepare:
 
 .PHONY: build_without_edgex
 build_without_edgex: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
@@ -54,16 +54,16 @@ pkg_without_edgex: build_without_edgex
 
 .PHONY: build_with_edgex
 build_with_edgex: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
 
 .PHONY: build_with_edgex_and_script
 build_with_edgex_and_script: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging script" -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "edgex include_nats_messaging script" -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
@@ -74,8 +74,8 @@ pkg_with_edgex: build_with_edgex
 
 .PHONY: build_with_fdb
 build_with_fdb: build_prepare
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "fdb" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "fdb" -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "fdb" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "fdb" -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
@@ -86,7 +86,7 @@ pkg_with_fdb: build_with_fdb
 
 .PHONY: build_core
 build_core: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags core -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags core -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiperd; fi
 	@mv ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@echo "Build successfully"
@@ -105,8 +105,8 @@ PLUGINS_IN_FULL := \
 .PHONY: build_full
 build_full: SHELL:=/bin/bash -euo pipefail
 build_full: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "full include_nats_messaging" -o kuiperd cmd/kuiperd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -o kuiper cmd/kuiper/main.go
+	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/v2/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/v2/cmd.LoadFileType=relative" -tags "full include_nats_messaging" -o kuiperd cmd/kuiperd/main.go
 	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
 	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
 	@while read plugin; do \
@@ -141,15 +141,6 @@ real_pkg:
 	@mv $(BUILD_PATH)/$(PACKAGE_NAME).zip $(BUILD_PATH)/$(PACKAGE_NAME).tar.gz $(PACKAGES_PATH)
 	@echo "Package build success"
 
-.PHONY: build_with_wasm
-build_with_wasm: build_prepare
-	GO111MODULE=on CGO_ENABLED=0 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -o kuiper cmd/kuiper/main.go
-	GO111MODULE=on CGO_ENABLED=1 go build -trimpath -ldflags="-s -w -X github.com/lf-edge/ekuiper/cmd.Version=$(VERSION) -X github.com/lf-edge/ekuiper/cmd.LoadFileType=relative" -tags "wasmedge" -o kuiperd cmd/kuiperd/main.go
-	@if [ "$$(uname -s)" = "Linux" ] && [ ! -z $$(which upx) ]; then upx ./kuiper; upx ./kuiperd; fi
-	@mv ./kuiper ./kuiperd $(BUILD_PATH)/$(PACKAGE_NAME)/bin
-	@echo "Build successfully"
-
-
 .PHONY: docker
 docker:
 	docker buildx build --no-cache --platform=linux/amd64 -t $(TARGET):$(VERSION) -f deploy/docker/Dockerfile . --load
diff --git a/contract/api/func.go b/contract/api/func.go
index 8b2970bb..9d999f43 100644
--- a/contract/api/func.go
+++ b/contract/api/func.go
@@ -21,10 +21,10 @@ type FunctionContext interface {
 
 type Function interface {
 	// Validate The argument is a list of xsql.Expr
-	Validate(args []interface{}) error
+	Validate(args []any) error
 	// Exec Execute the function, return the result and if execution is successful.
 	// If execution fails, return the error and false.
-	Exec(args []interface{}, ctx FunctionContext) (interface{}, bool)
+	Exec(ctx FunctionContext, args []any) (interface{}, bool)
 	// IsAggregate If this function is an aggregate function. Each parameter of an aggregate function will be a slice
 	IsAggregate() bool
 }
diff --git a/extension/function/accumulateWordCount/accumulateWordCount.go b/extension/function/accumulateWordCount/accumulateWordCount.go
index e8e09dc0..56abf2f6 100644
--- a/extension/function/accumulateWordCount/accumulateWordCount.go
+++ b/extension/function/accumulateWordCount/accumulateWordCount.go
@@ -44,7 +44,7 @@ func (f *accumulateWordCountFunc) Validate(args []any) error {
 	return nil
 }
 
-func (f *accumulateWordCountFunc) Exec(args []any, ctx api.FunctionContext) (any, bool) {
+func (f *accumulateWordCountFunc) Exec(ctx api.FunctionContext, args []any) (any, bool) {
 	logger := ctx.GetLogger()
 	fmt.Printf("Exec accumulate")
 	col, ok := args[0].(string)
diff --git a/extension/function/countPlusOne/countPlusOne.go b/extension/function/countPlusOne/countPlusOne.go
index d247f721..ff0a72f3 100644
--- a/extension/function/countPlusOne/countPlusOne.go
+++ b/extension/function/countPlusOne/countPlusOne.go
@@ -29,7 +29,7 @@ func (f *countPlusOneFunc) Validate(args []any) error {
 	return nil
 }
 
-func (f *countPlusOneFunc) Exec(args []any, ctx api.FunctionContext) (any, bool) {
+func (f *countPlusOneFunc) Exec(_ api.FunctionContext, args []any) (any, bool) {
 	arg, ok := args[0].([]interface{})
 	if !ok {
 		return fmt.Errorf("arg is not a slice, got %v", args[0]), false
diff --git a/extension/function/echo/echo.go b/extension/function/echo/echo.go
index a0687c0b..434f0005 100644
--- a/extension/function/echo/echo.go
+++ b/extension/function/echo/echo.go
@@ -29,7 +29,7 @@ func (f *echo) Validate(args []any) error {
 	return nil
 }
 
-func (f *echo) Exec(args []any, ctx api.FunctionContext) (any, bool) {
+func (f *echo) Exec(_ api.FunctionContext, args []any) (any, bool) {
 	result := args[0]
 	return result, true
 }
diff --git a/internal/binder/function/funcs_trans.go b/internal/binder/function/funcs_trans.go
index f8c966fb..0976b6e4 100644
--- a/internal/binder/function/funcs_trans.go
+++ b/internal/binder/function/funcs_trans.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -29,7 +29,7 @@ type compressFunc struct {
 	compressor   message.Compressor
 }
 
-func (c *compressFunc) Validate(args []interface{}) error {
+func (c *compressFunc) Validate(args []any) error {
 	var eargs []ast.Expr
 	for _, arg := range args {
 		if t, ok := arg.(ast.Expr); ok {
@@ -42,7 +42,7 @@ func (c *compressFunc) Validate(args []interface{}) error {
 	return ValidateTwoStrArg(nil, eargs)
 }
 
-func (c *compressFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+func (c *compressFunc) Exec(ctx api.FunctionContext, args []any) (any, bool) {
 	if args[0] == nil {
 		return nil, true
 	}
@@ -79,7 +79,7 @@ type decompressFunc struct {
 	decompressor message.Decompressor
 }
 
-func (d *decompressFunc) Validate(args []interface{}) error {
+func (d *decompressFunc) Validate(args []any) error {
 	var eargs []ast.Expr
 	for _, arg := range args {
 		if t, ok := arg.(ast.Expr); ok {
@@ -92,7 +92,7 @@ func (d *decompressFunc) Validate(args []interface{}) error {
 	return ValidateTwoStrArg(nil, eargs)
 }
 
-func (d *decompressFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+func (d *decompressFunc) Exec(ctx api.FunctionContext, args []any) (any, bool) {
 	if args[0] == nil {
 		return nil, true
 	}
diff --git a/internal/binder/function/static_executor.go b/internal/binder/function/static_executor.go
index 4f3fed62..1ada782d 100644
--- a/internal/binder/function/static_executor.go
+++ b/internal/binder/function/static_executor.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -46,7 +46,7 @@ func (f *funcExecutor) Validate(_ []interface{}) error {
 	return fmt.Errorf("unknow name")
 }
 
-func (f *funcExecutor) Exec(_ []interface{}, _ api.FunctionContext) (interface{}, bool) {
+func (f *funcExecutor) Exec(ctx api.FunctionContext, args []any) (interface{}, bool) {
 	return fmt.Errorf("unknow name"), false
 }
 
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index 26d0b6f9..12735b83 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -18,6 +18,7 @@ import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/lf-edge/ekuiper/v2/internal/io/file"
 	"github.com/lf-edge/ekuiper/v2/internal/io/http"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
 	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
@@ -30,7 +31,7 @@ import (
 func init() {
 	modules.RegisterSource("mqtt", mqtt.GetSource)
 	modules.RegisterSource("httppull", func() api.Source { return &http.HttpPullSource{} })
-	// modules.RegisterSource("httppush", func() api.Source { return &http.PushSource{} })
+	modules.RegisterSource("httppush", func() api.Source { return &http.HttpPushSource{} })
 	modules.RegisterSource("file", file.GetSource)
 	modules.RegisterSource("memory", func() api.Source { return memory.GetSource() })
 	modules.RegisterSource("neuron", neuron.GetSource)
@@ -52,6 +53,7 @@ func init() {
 
 	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
 	modules.RegisterConnection("nng", nng.CreateConnection)
+	modules.RegisterConnection("httppush", httpserver.CreateConnection)
 }
 
 type Manager struct{}
diff --git a/internal/binder/mock/mock_factory.go b/internal/binder/mock/mock_factory.go
index b66dc005..9355f388 100644
--- a/internal/binder/mock/mock_factory.go
+++ b/internal/binder/mock/mock_factory.go
@@ -73,7 +73,7 @@ func (m *mockFunc) Validate(_ []interface{}) error {
 	return nil
 }
 
-func (m *mockFunc) Exec(_ []interface{}, _ api.FunctionContext) (interface{}, bool) {
+func (m *mockFunc) Exec(ctx api.FunctionContext, args []any) (interface{}, bool) {
 	return nil, true
 }
 
diff --git a/internal/conf/conf.go b/internal/conf/conf.go
index 7b853601..4065f807 100644
--- a/internal/conf/conf.go
+++ b/internal/conf/conf.go
@@ -52,7 +52,7 @@ var (
 	TestId    string
 )
 
-type tlsConf struct {
+type TlsConf struct {
 	Certfile string `yaml:"certfile"`
 	Keyfile  string `yaml:"keyfile"`
 }
@@ -123,7 +123,7 @@ func (sc *SinkConf) Validate() error {
 type SourceConf struct {
 	HttpServerIp   string   `json:"httpServerIp" yaml:"httpServerIp"`
 	HttpServerPort int      `json:"httpServerPort" yaml:"httpServerPort"`
-	HttpServerTls  *tlsConf `json:"httpServerTls" yaml:"httpServerTls"`
+	HttpServerTls  *TlsConf `json:"httpServerTls" yaml:"httpServerTls"`
 }
 
 func (sc *SourceConf) Validate() error {
@@ -184,7 +184,7 @@ type KuiperConf struct {
 		Port                    int               `yaml:"port"`
 		RestIp                  string            `yaml:"restIp"`
 		RestPort                int               `yaml:"restPort"`
-		RestTls                 *tlsConf          `yaml:"restTls"`
+		RestTls                 *TlsConf          `yaml:"restTls"`
 		Prometheus              bool              `yaml:"prometheus"`
 		PrometheusPort          int               `yaml:"prometheusPort"`
 		PluginHosts             string            `yaml:"pluginHosts"`
diff --git a/internal/io/http/httppush_source.go b/internal/io/http/httppush_source.go
new file mode 100644
index 00000000..b80219b6
--- /dev/null
+++ b/internal/io/http/httppush_source.go
@@ -0,0 +1,110 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package http
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type HttpPushSource struct {
+	topic    string
+	sourceID string
+	ch       <-chan any
+	conf     *PushConf
+	props    map[string]any
+}
+
+type PushConf struct {
+	Method       string `json:"method"`
+	BufferLength int    `json:"bufferLength"`
+	DataSource   string `json:"datasource"`
+}
+
+func (h *HttpPushSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	cfg := &PushConf{
+		Method:       http.MethodPost,
+		BufferLength: 1024,
+	}
+	err := cast.MapToStruct(configs, cfg)
+	if err != nil {
+		return err
+	}
+	if cfg.Method != http.MethodPost && cfg.Method != http.MethodPut {
+		return fmt.Errorf("method %s is not supported, must be POST or PUT", cfg.Method)
+	}
+	if !strings.HasPrefix(cfg.DataSource, "/") {
+		return fmt.Errorf("property `endpoint` must start with /")
+	}
+
+	h.conf = cfg
+	h.props = configs
+	conf.Log.Debugf("Initialized with configurations %#v.", cfg)
+	return nil
+}
+
+func (h *HttpPushSource) Close(ctx api.StreamContext) error {
+	pubsub.CloseSourceConsumerChannel(h.topic, h.sourceID)
+	return connection.DetachConnection(ctx, h.conf.DataSource, h.props)
+}
+
+func (h *HttpPushSource) Connect(ctx api.StreamContext) error {
+	c, err := connection.FetchConnection(ctx, h.conf.DataSource, "httppush", h.props)
+	if err != nil {
+		return err
+	}
+	hc, ok := c.(*httpserver.HttpPushConnection)
+	if !ok {
+		return fmt.Errorf("connection isn't httppushConnection")
+	}
+	h.sourceID = fmt.Sprintf("%s_%s_%v", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
+	h.topic = hc.GetTopic()
+	return nil
+}
+
+func (h *HttpPushSource) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+	ch := pubsub.CreateSub(h.topic, nil, h.sourceID, 1024)
+	h.ch = ch
+	go func(ctx api.StreamContext) {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case v := <-h.ch:
+				data := v.([]byte)
+				e := infra.SafeRun(func() error {
+					ingest(ctx, data, nil, timex.GetNow())
+					return nil
+				})
+				if e != nil {
+					ingestError(ctx, e)
+				}
+			}
+		}
+	}(ctx)
+	return nil
+}
+
+var _ api.BytesSource = &HttpPushSource{}
diff --git a/internal/io/http/httpserver/data_server.go b/internal/io/http/httpserver/data_server.go
new file mode 100644
index 00000000..20dcc201
--- /dev/null
+++ b/internal/io/http/httpserver/data_server.go
@@ -0,0 +1,133 @@
+// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gorilla/handlers"
+	"github.com/gorilla/mux"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/memory/pubsub"
+	"github.com/lf-edge/ekuiper/v2/internal/topo/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type GlobalServerManager struct {
+	sync.RWMutex
+	endpoint map[string]string
+	server   *http.Server
+	router   *mux.Router
+}
+
+var manager *GlobalServerManager
+
+func InitGlobalServerManager(ip string, port int, tlsConf *conf.TlsConf) {
+	r := mux.NewRouter()
+	s := &http.Server{
+		Addr: cast.JoinHostPortInt(ip, port),
+		// Good practice to set timeouts to avoid Slowloris attacks.
+		WriteTimeout: time.Second * 60 * 5,
+		ReadTimeout:  time.Second * 60 * 5,
+		IdleTimeout:  time.Second * 60,
+		Handler:      handlers.CORS(handlers.AllowedHeaders([]string{"Accept", "Accept-Language", "Content-Type", "Content-Language", "Origin", "Authorization"}), handlers.AllowedMethods([]string{"POST", "GET", "PUT", "DELETE", "HEAD"}))(r),
+	}
+	manager = &GlobalServerManager{
+		endpoint: map[string]string{},
+		server:   s,
+		router:   r,
+	}
+	go func(m *GlobalServerManager) {
+		if tlsConf == nil {
+			s.ListenAndServe()
+		} else {
+			s.ListenAndServeTLS(conf.Config.Source.HttpServerTls.Certfile, conf.Config.Source.HttpServerTls.Keyfile)
+		}
+	}(manager)
+	time.Sleep(500 * time.Millisecond)
+}
+
+func ShutDown() {
+	manager.Shutdown()
+	manager = nil
+}
+
+func RegisterEndpoint(endpoint string, method string) (string, error) {
+	return manager.RegisterEndpoint(endpoint, method)
+}
+
+func UnregisterEndpoint(endpoint string) {
+	manager.UnregisterEndpoint(endpoint)
+}
+
+const (
+	TopicPrefix = "$$httppush/"
+)
+
+func (m *GlobalServerManager) RegisterEndpoint(endpoint string, method string) (string, error) {
+	var topic string
+	var ok bool
+	m.Lock()
+	topic, ok = m.endpoint[endpoint]
+	if ok {
+		return topic, nil
+	} else {
+		topic = TopicPrefix + endpoint
+		m.endpoint[endpoint] = topic
+	}
+	m.Unlock()
+	pubsub.CreatePub(topic)
+	m.router.HandleFunc(endpoint, func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+		data, err := io.ReadAll(r.Body)
+		if err != nil {
+			handleError(w, err, "Fail to decode data")
+			return
+		}
+		pubsub.ProduceAny(context.Background(), topic, data)
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	}).Methods(method)
+	return topic, nil
+}
+
+func (m *GlobalServerManager) UnregisterEndpoint(endpoint string) {
+	var ok bool
+	m.Lock()
+	_, ok = m.endpoint[endpoint]
+	if !ok {
+		return
+	}
+	delete(m.endpoint, endpoint)
+	m.Unlock()
+	pubsub.RemovePub(TopicPrefix + endpoint)
+}
+
+func (m *GlobalServerManager) Shutdown() {
+	m.server.Shutdown(context.Background())
+}
+
+func handleError(w http.ResponseWriter, err error, prefix string) {
+	message := prefix
+	if message != "" {
+		message += ": "
+	}
+	message += err.Error()
+	http.Error(w, message, http.StatusBadRequest)
+}
diff --git a/internal/io/http/httpserver/httpconn.go b/internal/io/http/httpserver/httpconn.go
new file mode 100644
index 00000000..644f1baa
--- /dev/null
+++ b/internal/io/http/httpserver/httpconn.go
@@ -0,0 +1,66 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+type HttpPushConnection struct {
+	topic    string
+	cfg      *connectionCfg
+	endpoint string
+	method   string
+}
+
+type connectionCfg struct {
+	Datasource string `json:"datasource"`
+	Method     string `json:"method"`
+}
+
+func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	cfg := &connectionCfg{}
+	if err := cast.MapToStruct(props, cfg); err != nil {
+		return nil, err
+	}
+	topic, err := RegisterEndpoint(cfg.Datasource, cfg.Method)
+	if err != nil {
+		return nil, err
+	}
+	return &HttpPushConnection{
+		topic:    topic,
+		cfg:      cfg,
+		endpoint: cfg.Datasource,
+		method:   cfg.Method,
+	}, nil
+}
+
+func (h *HttpPushConnection) Ping(ctx api.StreamContext) error {
+	return nil
+}
+
+func (h *HttpPushConnection) DetachSub(ctx api.StreamContext, props map[string]any) {
+	UnregisterEndpoint(h.endpoint)
+}
+
+func (h *HttpPushConnection) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (h *HttpPushConnection) GetTopic() string {
+	return h.topic
+}
diff --git a/internal/io/memory/pubsub/manager.go b/internal/io/memory/pubsub/manager.go
index 75b417db..b0d15993 100644
--- a/internal/io/memory/pubsub/manager.go
+++ b/internal/io/memory/pubsub/manager.go
@@ -109,6 +109,10 @@ func RemovePub(topic string) {
 	}
 }
 
+func ProduceAny(ctx api.StreamContext, topic string, data any) {
+	doProduce(ctx, topic, data)
+}
+
 func Produce(ctx api.StreamContext, topic string, data MemTuple) {
 	doProduce(ctx, topic, data)
 }
diff --git a/internal/plugin/js/function.go b/internal/plugin/js/function.go
index bdb766d1..ff66f994 100644
--- a/internal/plugin/js/function.go
+++ b/internal/plugin/js/function.go
@@ -61,7 +61,7 @@ func (f *JSFunc) Validate(_ []interface{}) error {
 	return nil
 }
 
-func (f *JSFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+func (f *JSFunc) Exec(ctx api.FunctionContext, args []any) (interface{}, bool) {
 	ctx.GetLogger().Debugf("running js func with args %+v", args)
 	if len(args) != len(f.args) {
 		f.args = make([]goja.Value, len(args))
diff --git a/internal/plugin/portable/runtime/function.go b/internal/plugin/portable/runtime/function.go
index 5cb6e3a1..7bb62765 100644
--- a/internal/plugin/portable/runtime/function.go
+++ b/internal/plugin/portable/runtime/function.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+// Copyright 2022-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -109,7 +109,7 @@ func (f *PortableFunc) Validate(args []interface{}) error {
 	}
 }
 
-func (f *PortableFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+func (f *PortableFunc) Exec(ctx api.FunctionContext, args []any) (interface{}, bool) {
 	if !f.referenced && f.onFirstExec != nil {
 		f.referenced = true
 		f.onFirstExec(ctx)
diff --git a/internal/server/server.go b/internal/server/server.go
index ec648f82..34847546 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -35,6 +35,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/binder/io"
 	"github.com/lf-edge/ekuiper/v2/internal/binder/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
 	"github.com/lf-edge/ekuiper/v2/internal/keyedstate"
 	meta2 "github.com/lf-edge/ekuiper/v2/internal/meta"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
@@ -160,6 +161,7 @@ func StartUp(Version string) {
 	keyedstate.InitKeyedStateKV()
 
 	meta2.InitYamlConfigManager()
+	httpserver.InitGlobalServerManager(conf.Config.Source.HttpServerIp, conf.Config.Source.HttpServerPort, conf.Config.Source.HttpServerTls)
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
diff --git a/internal/service/externalFunc.go b/internal/service/externalFunc.go
index 51e97e1d..069c30ce 100644
--- a/internal/service/externalFunc.go
+++ b/internal/service/externalFunc.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -27,7 +27,7 @@ func (f *ExternalFunc) Validate(_ []interface{}) error {
 	return nil
 }
 
-func (f *ExternalFunc) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
+func (f *ExternalFunc) Exec(ctx api.FunctionContext, args []any) (interface{}, bool) {
 	if r, err := f.exe.InvokeFunction(ctx, f.methodName, args); err != nil {
 		return err, false
 	} else {
diff --git a/internal/xsql/func_invoker.go b/internal/xsql/func_invoker.go
index a28505fe..55a18da6 100644
--- a/internal/xsql/func_invoker.go
+++ b/internal/xsql/func_invoker.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -47,7 +47,7 @@ func ExecFunc(funcName string, f api.Function, args []interface{}, fctx api.Func
 	if mf, ok := f.(MultiFunc); ok {
 		return mf.ExecWithName(args, fctx, funcName)
 	} else {
-		return f.Exec(args, fctx)
+		return f.Exec(fctx, args)
 	}
 }
 


Test patch:
diff --git a/internal/binder/function/funcs_trans_test.go b/internal/binder/function/funcs_trans_test.go
index 941be756..41ffc820 100644
--- a/internal/binder/function/funcs_trans_test.go
+++ b/internal/binder/function/funcs_trans_test.go
@@ -1,4 +1,4 @@
-// Copyright 2023 EMQ Technologies Co., Ltd.
+// Copyright 2023-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -72,7 +72,7 @@ func TestCompressExec(t *testing.T) {
 		},
 	}
 	for i, tt := range tests {
-		result, _ := f.Exec(tt.args, fctx)
+		result, _ := f.Exec(fctx, tt.args)
 		if !reflect.DeepEqual(result, tt.result) {
 			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
 		}
@@ -126,7 +126,7 @@ func TestDecompressExec(t *testing.T) {
 		},
 	}
 	for i, tt := range tests {
-		result, _ := f.Exec(tt.args, fctx)
+		result, _ := f.Exec(fctx, tt.args)
 		if !reflect.DeepEqual(result, tt.result) {
 			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
 		}
diff --git a/internal/conf/conf_test.go b/internal/conf/conf_test.go
index fabd1621..32fdb116 100644
--- a/internal/conf/conf_test.go
+++ b/internal/conf/conf_test.go
@@ -60,7 +60,7 @@ func TestSourceConfValidate(t *testing.T) {
 		}, {
 			s: &SourceConf{
 				HttpServerPort: 9090,
-				HttpServerTls: &tlsConf{
+				HttpServerTls: &TlsConf{
 					Certfile: "certfile",
 					Keyfile:  "keyfile",
 				},
@@ -68,7 +68,7 @@ func TestSourceConfValidate(t *testing.T) {
 			e: &SourceConf{
 				HttpServerIp:   "0.0.0.0",
 				HttpServerPort: 9090,
-				HttpServerTls: &tlsConf{
+				HttpServerTls: &TlsConf{
 					Certfile: "certfile",
 					Keyfile:  "keyfile",
 				},
diff --git a/internal/io/http/httppush_source_test.go b/internal/io/http/httppush_source_test.go
new file mode 100644
index 00000000..221093a3
--- /dev/null
+++ b/internal/io/http/httppush_source_test.go
@@ -0,0 +1,72 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package http
+
+import (
+	"encoding/hex"
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/io/http/httpserver"
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterConnection("httppush", httpserver.CreateConnection)
+}
+
+func TestHttpPushSource(t *testing.T) {
+	connection.InitConnectionManager4Test()
+	ip := "127.0.0.1"
+	port := 10081
+	httpserver.InitGlobalServerManager(ip, port, nil)
+	defer httpserver.ShutDown()
+	ctx := mockContext.NewMockContext("1", "2")
+	s := &HttpPushSource{}
+	require.NoError(t, s.Provision(ctx, map[string]any{
+		"method":     "POST",
+		"datasource": "/post",
+	}))
+	require.NoError(t, s.Connect(ctx))
+	recvData := make(chan []byte, 10)
+	require.NoError(t, s.Subscribe(ctx, func(ctx api.StreamContext, data []byte, meta map[string]any, ts time.Time) {
+		recvData <- data
+	}, func(ctx api.StreamContext, err error) {}))
+	require.NoError(t, testx.TestHttp(&http.Client{}, fmt.Sprintf("http://%v:%v/post", ip, port), "POST"))
+	x := <-recvData
+	require.Equal(t, `7b0a2020202020202020227469746c65223a2022506f7374207469746c65222c0a202020202020202022626f6479223a2022506f7374206465736372697074696f6e222c0a202020202020202022757365724964223a20310a202020207d`, hex.EncodeToString(x))
+	require.NoError(t, s.Close(ctx))
+}
+
+func TestHttpPushProvisionErr(t *testing.T) {
+	ctx := mockContext.NewMockContext("1", "2")
+	s := &HttpPushSource{}
+	require.Error(t, s.Provision(ctx, map[string]any{
+		"method":     "GET",
+		"datasource": "/post",
+	}))
+	require.Error(t, s.Provision(ctx, map[string]any{
+		"method":     "POST",
+		"datasource": "post",
+	}))
+}
diff --git a/internal/io/http/httpserver/data_server_test.go b/internal/io/http/httpserver/data_server_test.go
new file mode 100644
index 00000000..766d5b57
--- /dev/null
+++ b/internal/io/http/httpserver/data_server_test.go
@@ -0,0 +1,75 @@
+// Copyright 2022-2023 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+)
+
+func TestEndpoints(t *testing.T) {
+	ip := "127.0.0.1"
+	port := 10082
+	InitGlobalServerManager(ip, port, nil)
+	defer ShutDown()
+	endpoints := []string{
+		"/ee1", "/eb2", "/ec3",
+	}
+	RegisterEndpoint(endpoints[0], "POST")
+	RegisterEndpoint(endpoints[1], "PUT")
+	RegisterEndpoint(endpoints[2], "POST")
+	require.Equal(t, map[string]struct{}{
+		"/ee1": {}, "/eb2": {}, "/ec3": {},
+	}, GetEndpoints())
+	UnregisterEndpoint(endpoints[0])
+	UnregisterEndpoint(endpoints[1])
+	UnregisterEndpoint(endpoints[2])
+	require.Equal(t, map[string]struct{}{}, GetEndpoints())
+
+	urlPrefix := fmt.Sprintf("http://%v:%v", ip, port)
+	client := &http.Client{}
+	RegisterEndpoint(endpoints[0], "POST")
+	RegisterEndpoint(endpoints[1], "PUT")
+	var err error
+	// wait for http server start
+	for i := 0; i < 3; i++ {
+		err = testx.TestHttp(client, urlPrefix+endpoints[1], "PUT")
+		if err == nil {
+			break
+		}
+		time.Sleep(time.Millisecond * 500)
+	}
+	require.NoError(t, err)
+}
+
+func GetEndpoints() map[string]struct{} {
+	return manager.GetEndpoints()
+}
+
+func (m *GlobalServerManager) GetEndpoints() map[string]struct{} {
+	ma := make(map[string]struct{})
+	m.Lock()
+	defer m.Unlock()
+	for k := range m.endpoint {
+		ma[k] = struct{}{}
+	}
+	return ma
+}
diff --git a/internal/io/http/httpserver/httpconn_test.go b/internal/io/http/httpserver/httpconn_test.go
new file mode 100644
index 00000000..693a8565
--- /dev/null
+++ b/internal/io/http/httpserver/httpconn_test.go
@@ -0,0 +1,40 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package httpserver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestHttpConn(t *testing.T) {
+	ip := "127.0.0.1"
+	port := 10084
+	InitGlobalServerManager(ip, port, nil)
+	defer ShutDown()
+	ctx := mockContext.NewMockContext("1", "2")
+	props := map[string]any{
+		"datasource": "/post",
+		"method":     "POST",
+	}
+	c, err := CreateConnection(ctx, props)
+	require.NoError(t, err)
+	require.NoError(t, c.Ping(ctx))
+	c.DetachSub(ctx, props)
+	require.NoError(t, c.Close(ctx))
+}
diff --git a/internal/plugin/js/function_test.go b/internal/plugin/js/function_test.go
index fe920224..b5db91b6 100644
--- a/internal/plugin/js/function_test.go
+++ b/internal/plugin/js/function_test.go
@@ -99,7 +99,7 @@ func TestScalarFuncHappyPath(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result, _ := ff.Exec(tt.args, fctx)
+			result, _ := ff.Exec(fctx, tt.args)
 			assert.Equal(t, tt.want, result)
 		})
 	}
@@ -155,7 +155,7 @@ func TestAggFuncHappyPath(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result, _ := ff.Exec(tt.args, fctx)
+			result, _ := ff.Exec(fctx, tt.args)
 			assert.Equal(t, tt.want, result)
 		})
 	}
@@ -222,7 +222,7 @@ func TestScalarFuncComplexType(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result, _ := ff.Exec(tt.args, fctx)
+			result, _ := ff.Exec(fctx, tt.args)
 			assert.Equal(t, tt.want, result)
 		})
 	}
diff --git a/internal/testx/testUtil.go b/internal/testx/testUtil.go
index ed38c696..6e7cbced 100644
--- a/internal/testx/testUtil.go
+++ b/internal/testx/testUtil.go
@@ -15,8 +15,11 @@
 package testx
 
 import (
+	"bytes"
 	"context"
+	"fmt"
 	"log"
+	"net/http"
 	"sync"
 
 	mqtt "github.com/mochi-mqtt/server/v2"
@@ -82,3 +85,24 @@ func InitBroker(id string) (string, func(), error) {
 		wg.Wait()
 	}, nil
 }
+
+func TestHttp(client *http.Client, url string, method string) error {
+	r, err := http.NewRequest(method, url, bytes.NewBuffer(body))
+	if err != nil {
+		return err
+	}
+	resp, err := client.Do(r)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("status code is not 200 for %s, code:%v", url, resp.StatusCode)
+	}
+	return nil
+}
+
+var body = []byte(`{
+        "title": "Post title",
+        "body": "Post description",
+        "userId": 1
+    }`)