Repository: lf-edge/ekuiper
Pull request #3051 (2024-07-24 08:43:33)

feat: support httppull lookup

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?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 `80.00000%` with `11 lines` in your changes missing coverage. Please review.
> Project coverage is 67.91%. Comparing base [(`c300688`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/c3006889c623688b76d55680cb509863052b4f33?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`c07df38`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/c07df38f6fe8cac0c52249680e1f348d2b668402?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).
> Report is 11 commits behind head on master.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?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/io/http/lookup\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?src=pr&el=tree&filepath=internal%2Fio%2Fhttp%2Flookup_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vaHR0cC9sb29rdXBfc291cmNlLmdv) | 78.57% | [6 Missing and 3 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/http/httppull\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?src=pr&el=tree&filepath=internal%2Fio%2Fhttp%2Fhttppull_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vaHR0cC9odHRwcHVsbF9zb3VyY2UuZ28=) | 83.33% | [2 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?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    #3051      +/-   ##
==========================================
+ Coverage   67.64%   67.91%   +0.28%     
==========================================
  Files         357      389      +32     
  Lines       34952    37028    +2076     
==========================================
+ Hits        23640    25147    +1507     
- Misses       9189     9608     +419     
- Partials     2123     2273     +150     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3051?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/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index b2c8be2c..71974bcc 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -51,7 +51,7 @@ func init() {
 	modules.RegisterSink("websocket", func() api.Sink { return websocket.GetSink() })
 
 	modules.RegisterLookupSource("memory", memory.GetLookupSource)
-	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
+	modules.RegisterLookupSource("httppull", http.GetLookUpSource)
 
 	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
 	modules.RegisterConnection("nng", nng.CreateConnection)
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index f42cbefd..3e21dea5 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -60,20 +60,28 @@ func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]a
 }
 
 func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, error) {
-	headers, err := hps.parseHeaders(ctx, hps.tokens)
+	result, latestMD5, err := doPull(ctx, hps.ClientConf, hps.lastMD5)
 	if err != nil {
 		return nil, err
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, headers, true, []byte(hps.config.Body))
+	hps.lastMD5 = latestMD5
+	return result, nil
+}
+
+func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]any, string, error) {
+	headers, err := c.parseHeaders(ctx, c.tokens)
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
-	results, newMD5, err := hps.parseResponse(ctx, resp, hps.lastMD5)
+	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, true, []byte(c.config.Body))
 	if err != nil {
-		return nil, err
+		return nil, "", err
+	}
+	results, newMD5, err := c.parseResponse(ctx, resp, lastMD5)
+	if err != nil {
+		return nil, "", err
 	}
-	hps.lastMD5 = newMD5
-	return results, nil
+	return results, newMD5, nil
 }
 
 func GetSource() api.Source {
diff --git a/internal/io/http/lookup_source.go b/internal/io/http/lookup_source.go
new file mode 100644
index 00000000..d4f5ddd7
--- /dev/null
+++ b/internal/io/http/lookup_source.go
@@ -0,0 +1,96 @@
+// 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 (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type HttpLookupSource struct {
+	*ClientConf
+}
+
+func (hls *HttpLookupSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	pc := &pullSourceConfig{}
+	if err := cast.MapToStruct(configs, pc); err != nil {
+		return err
+	}
+	if hls.ClientConf == nil {
+		hls.ClientConf = &ClientConf{}
+	}
+	return hls.InitConf(pc.Path, configs)
+}
+
+func (hls *HttpLookupSource) Close(ctx api.StreamContext) error {
+	return nil
+}
+
+func (hls *HttpLookupSource) Connect(ctx api.StreamContext) error {
+	return nil
+}
+
+func (hls *HttpLookupSource) Lookup(ctx api.StreamContext, fields []string, keys []string, values []any) ([]map[string]any, error) {
+	resps, _, err := doPull(ctx, hls.ClientConf, "")
+	if err != nil {
+		return nil, err
+	}
+	resps = pruneData(resps, fields)
+	return lookupJoin(resps, keys, values), nil
+}
+
+func pruneData(data []map[string]any, fields []string) []map[string]any {
+	for index, row := range data {
+		for key := range row {
+			if !findColumn(key, fields) {
+				delete(row, key)
+			}
+		}
+		data[index] = row
+	}
+	return data
+}
+
+func findColumn(column string, fields []string) bool {
+	for _, field := range fields {
+		if field == column {
+			return true
+		}
+	}
+	return false
+}
+
+func lookupJoin(dataMap []map[string]interface{}, keys []string, values []interface{}) []map[string]any {
+	var resps []map[string]interface{}
+	for _, resp := range dataMap {
+		match := true
+		for i, k := range keys {
+			if val, ok := resp[k]; !ok || val != values[i] {
+				match = false
+				break
+			}
+		}
+		if match {
+			resps = append(resps, resp)
+		}
+	}
+	return resps
+}
+
+func GetLookUpSource() api.Source {
+	return &HttpLookupSource{}
+}
+
+var _ api.LookupSource = &HttpLookupSource{}


Test patch:
diff --git a/internal/io/http/lookup_source_test.go b/internal/io/http/lookup_source_test.go
new file mode 100644
index 00000000..c0f86ea1
--- /dev/null
+++ b/internal/io/http/lookup_source_test.go
@@ -0,0 +1,46 @@
+// 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 (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestLookup(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	hls := &HttpLookupSource{}
+	require.NoError(t, hls.Provision(ctx, map[string]any{
+		"url":        server.URL,
+		"datasource": "/get",
+		"method":     "get",
+	}))
+	require.NoError(t, hls.Connect(ctx))
+	got, err := hls.Lookup(ctx, []string{"code"}, []string{"code"}, []any{float64(200)})
+	require.NoError(t, err)
+	require.Equal(t, []map[string]any{
+		{
+			"code": float64(200),
+		},
+	}, got)
+	require.NoError(t, hls.Close(ctx))
+}