feat(io): support httppull state (#3804)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

docs/en_US/guide/sources/builtin/http_pull.md

@@ -38,6 +38,8 @@ default:
   headers:
     Accept: application/json
   # how to check the response status, by status code or by body
+  states:
+  #   state_key: state_value  
   responseType: code
   #  # Get token
 #  oAuth:
@@ -82,6 +84,7 @@ Use can specify the global HTTP pull settings here. The configuration items spec
 - `body`: The body of request, such as `'{"data": "data", "method": 1}'`
 - `bodyType`: Body type, it could be none|text|json|html|xml|javascript|format.
 - `headers`: The HTTP request headers that you want to send along with the HTTP request.
+- `states`: The status of httppull, can update the url by rendering.
 - `responseType`: Define how to parse the HTTP response. There are two types defined:
   - `code`: To check the response status from the HTTP status code.
   - `body`: To check the response status from the response body. The body must be "application/json" content type and contains a "code" field.
@@ -131,6 +134,12 @@ The following configurations are designed under the assumption that the authenti
 
 `incremental`: If it's set to `true`, then will compare with the last result; If the responses of two requests are the same, then will skip sending out the result.
 
+#### State Update
+
+State updates are dynamically updated at runtime. When creating the http_pull source, you can specify the initial state. The state can currently be rendered in the HTTP request URL.  The format for these properties is based on the [data template](../../sinks/data_template.md) syntax.
+
+State can also be updated based on the results of the http_pull result. When QOS is set to 1, the state will be periodically flushed to disk and loaded after the next boot.
+
 #### Dynamic Properties
 
 Dynamic properties adapt in real time and can be employed to customize the HTTP request's URL, body, and header. The format for these properties is based on the [data template](../../sinks/data_template.md) syntax.

docs/zh_CN/guide/sources/builtin/http_pull.md

@@ -38,6 +38,8 @@ default:
   headers:
     Accept: application/json
   # 如何检查响应状态，支持通过状态码或 body
+  states:
+ #   state_key: state_value  
   responseType: code
   # 获取 token
 #  oAuth:
@@ -82,6 +84,7 @@ application_conf: #Conf_key
 - `body`：请求的正文，例如 `{"data": "data", "method": 1}`
 - `bodyType`：正文类型，可选值 none、text、json、html、xml、javascript、format.
 - `headers`：需要与 HTTP 请求一起发送的 HTTP 请求标头。
+- `states`: httppull 的参数状态，可以通过渲染的方式更新 url。
 - `responseType`：定义如何解析 HTTP 响应。目前支持两种方式：
   - `code`：通过 HTTP 响应码判断响应状态。
   - `body`：通过 HTTP 响应正文判断响应状态。要求响应正文为 JSON 格式且其中包含 code 字段。
@@ -130,6 +133,12 @@ OAuth 2.0 是一个授权协议，让 API 客户端有限度地访问网络服
 
 `incremental`：如设置为 `true`，则将与上次的结果进行比较；如果两次请求的响应相同，则将跳过发送结果。
 
+#### 状态更新
+
+状态更新是指在运行时会动态更新的状态，在创建 http_pull 时可以指定初始状态。状态目前可以渲染在 HTTP 请求的 URL 中，其语法基于[数据模板](../../sinks/data_template.md)格式的动态属性。
+
+并且状态可以根据 http_pull 的结果进行更新。当开启了 QOS 为 1 后，状态将会被定期落盘，并在下一次启动后加载。
+
 #### 动态属性
 
 动态属性是指在运行时会动态更新的属性。您可以使用动态属性来指定 HTTP 请求的 URL、正文和标头。其语法基于[数据模板](../../sinks/data_template.md)格式的动态属性。

internal/io/http/httppull_source.go

@@ -26,6 +26,28 @@ import (
 type HttpPullSource struct {
 	*ClientConf
 	lastMD5 string
+	psc     *pullSourceConfig
+}
+
+func (hps *HttpPullSource) GetOffset() (any, error) {
+	return hps.psc.States, nil
+}
+
+func (hps *HttpPullSource) Rewind(offset any) error {
+	m, ok := offset.(map[string]interface{})
+	if ok {
+		for k, v := range m {
+			hps.psc.States[k] = v
+		}
+	}
+	return nil
+}
+
+func (hps *HttpPullSource) ResetOffset(input map[string]any) error {
+	for k, v := range input {
+		hps.psc.States[k] = v
+	}
+	return nil
 }
 
 func (hps *HttpPullSource) Pull(ctx api.StreamContext, trigger time.Time, ingest api.TupleIngest, ingestError api.ErrorIngest) {
@@ -47,30 +69,32 @@ func (hps *HttpPullSource) Connect(ctx api.StreamContext, sch api.StatusChangeHa
 }
 
 type pullSourceConfig struct {
-	Path string `json:"datasource"`
+	Path   string         `json:"datasource"`
+	States map[string]any `json:"states"`
 }
 
 func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]any) error {
-	pc := &pullSourceConfig{}
+	pc := &pullSourceConfig{States: map[string]any{}}
 	if err := cast.MapToStruct(configs, pc); err != nil {
 		return err
 	}
 	if hps.ClientConf == nil {
 		hps.ClientConf = &ClientConf{}
 	}
+	hps.psc = pc
 	return hps.InitConf(ctx, pc.Path, configs)
 }
 
 func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, error) {
-	result, latestMD5, err := doPull(ctx, hps.ClientConf, hps.lastMD5)
+	result, latestMD5, err := hps.doPullInternal(ctx, hps.ClientConf, hps.lastMD5)
 	if err != nil {
 		return nil, err
 	}
 	hps.lastMD5 = latestMD5
 	return result, nil
 }
 
-func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]any, string, error) {
+func (hps *HttpPullSource) doPullInternal(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
@@ -79,7 +103,14 @@ func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]
 	if err != nil {
 		return nil, "", err
 	}
-	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, []byte(newBody))
+	newUrl := c.config.Url
+	if len(hps.psc.States) > 0 {
+		newUrl, err = ctx.ParseTemplate(c.config.Url, hps.psc.States)
+		if err != nil {
+			return nil, "", err
+		}
+	}
+	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, newUrl, headers, []byte(newBody))
 	if err != nil {
 		return nil, "", err
 	}
@@ -88,9 +119,21 @@ func doPull(ctx api.StreamContext, c *ClientConf, lastMD5 string) ([]map[string]
 	if err != nil {
 		return nil, "", err
 	}
+	hps.updateState(results)
 	return results, newMD5, nil
 }
 
+func (hps *HttpPullSource) updateState(results []map[string]interface{}) {
+	for _, r := range results {
+		for k, v := range r {
+			_, ok := hps.psc.States[k]
+			if ok {
+				hps.psc.States[k] = v
+			}
+		}
+	}
+}
+
 func GetSource() api.Source {
 	return &HttpPullSource{}
 }

internal/io/http/httppull_source_test.go

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"testing"
 	"time"
 
@@ -90,6 +91,14 @@ func handleRefresh(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(resp)
 }
 
+func handleParam(w http.ResponseWriter, r *http.Request) {
+	v, _ := strconv.ParseInt(r.URL.Query().Get("a"), 10, 64)
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusOK)
+	v = v + 1
+	json.NewEncoder(w).Encode(map[string]interface{}{"a": v})
+}
+
 func createServer() *httptest.Server {
 	router := http.NewServeMux()
 	router.HandleFunc("/get", handleGet)
@@ -98,10 +107,65 @@ func createServer() *httptest.Server {
 	router.HandleFunc("/codeErr", handleCodeErr)
 	router.HandleFunc("/auth", handleAuth)
 	router.HandleFunc("/refresh", handleRefresh)
+	router.HandleFunc("/param", handleParam)
 	server := httptest.NewServer(router)
 	return server
 }
 
+func TestHttpPullStateRewind(t *testing.T) {
+	ctx := mockContext.NewMockContext("1", "2")
+	require.NotNil(t, GetSource())
+	source := &HttpPullSource{}
+	require.NoError(t, source.Provision(ctx, map[string]any{
+		"url":        "http://www.mock.com",
+		"datasource": "/dd",
+		"method":     "get",
+		"states":     map[string]interface{}{"a": 1},
+	}))
+	v, err := source.GetOffset()
+	require.NoError(t, err)
+	require.NotNil(t, v)
+	require.NoError(t, source.Rewind(map[string]any{"a": 1}))
+	require.NoError(t, source.ResetOffset(map[string]any{"a": 1}))
+}
+
+func TestHttpPullStateSource(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+	ctx := mockContext.NewMockContext("1", "2")
+	source := &HttpPullSource{}
+	require.NoError(t, source.Provision(ctx, map[string]any{
+		"url":        server.URL,
+		"datasource": "/param?a={{.a}}",
+		"method":     "get",
+		"states":     map[string]interface{}{"a": 1},
+	}))
+	require.NoError(t, source.Connect(ctx, func(status string, message string) {
+		// do nothing
+	}))
+	dataCh := make(chan any, 1)
+	source.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		dataCh <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Equal(t, []map[string]interface{}{
+		{
+			"a": float64(2),
+		},
+	}, <-dataCh)
+	source.Pull(ctx, time.Now(), func(ctx api.StreamContext, data any, meta map[string]any, ts time.Time) {
+		dataCh <- data
+	}, func(ctx api.StreamContext, err error) {})
+	require.Equal(t, []map[string]interface{}{
+		{
+			"a": float64(3),
+		},
+	}, <-dataCh)
+	source.Close(ctx)
+	close(dataCh)
+}
+
 func TestHttpPullSource(t *testing.T) {
 	server := createServer()
 	defer func() {

internal/io/http/lookup_source.go

@@ -17,6 +17,7 @@ package http
 import (
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 
+	"github.com/lf-edge/ekuiper/v2/internal/pkg/httpx"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 )
 
@@ -96,3 +97,24 @@ func GetLookUpSource() api.Source {
 }
 
 var _ api.LookupSource = &HttpLookupSource{}
+
+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
+	}
+	newBody, err := ctx.ParseTemplate(c.config.Body, c.tokens)
+	if err != nil {
+		return nil, "", err
+	}
+	resp, err := httpx.Send(ctx.GetLogger(), c.client, c.config.BodyType, c.config.Method, c.config.Url, headers, []byte(newBody))
+	if err != nil {
+		return nil, "", err
+	}
+	defer resp.Body.Close()
+	results, newMD5, err := c.parseResponse(ctx, resp, lastMD5, true, false)
+	if err != nil {
+		return nil, "", err
+	}
+	return results, newMD5, nil
+}