feat: add httpsqs client

Author: royee

.gitignore

@@ -0,0 +1,19 @@
+### Go template
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+### IDE
+.idea

LICENSE.md

@@ -0,0 +1,9 @@
+MIT License
+
+Copyright (c) 2025 gomooth,save95
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,94 @@
+# httpsqs
+
+HTTPSQS 队列客户端
+
+[基于HTTP协议的轻量级开源简单队列服务：HTTPSQS@张宴](http://blog.zyan.cc/httpsqs/)
+
+```golang
+
+type IClient interface {
+    // Put 入队列（将文本消息放入队列）
+    // 返回：当前队列的读取位置点 pos，及可能存在的错误
+    // pos = -1 则表示队列已满，同时也会返回 err
+    Put(ctx context.Context, name, data string) (pos int64, err error)
+    
+    // Get 出队列（从队列中取出文本消息）
+    // 返回 文本消息 data，当前队列的读取位置点 pos，及可能存在的错误
+    // pos = -1 则表示没有未取出的消息队列（HTTPSQS_GET_END）
+    Get(ctx context.Context, name string) (data string, pos int64, err error)
+    
+    // Status 查看队列状态
+    Status(ctx context.Context, name string) (*Status, error)
+    
+    // View 查看指定队列位置点的内容
+    // 跟一般的队列系统不同的是，HTTPSQS 可以查看指定队列ID（队列点）的内容，包括未出、已出的队列内容。
+    // 可以方便地观测进入队列的内容是否正确。另外，假设有一个发送手机短信的队列，由客户端守护进程从队列
+    // 中取出信息，并调用“短信网关接口”发送短信。但是，如果某段时间“短信网关接口”有故障，而这段时间队列
+    // 位置点300~900的信息已经出队列，但是发送短信失败，我们还可以在位置点300~900被覆盖前，查看到这些
+    // 位置点的内容，作相应的处理。
+    View(ctx context.Context, name string, pos int64) (string, error)
+    
+    // Reset 重置指定队列
+    Reset(ctx context.Context, name string) error
+    
+    // SetMaxQueue 更改指定队列的最大队列数量。默认的最大队列长度（100万条）
+    SetMaxQueue(ctx context.Context, name string, max int) error
+    
+    // SetSyncTime 不停止服务的情况下，修改定时刷新内存缓冲区内容到磁盘的间隔时间
+    // 从HTTPSQS 1.3版本开始支持此功能。
+    // 默认间隔时间：5秒 或 httpsqs -s <second> 参数设置的值。
+    SetSyncTime(ctx context.Context, name string, duration time.Duration) error
+}
+
+```
+
+## Usage
+
+### Start using it
+Download and install it:
+
+```shell
+go get github.com/gomooth/httpsqs
+
+```
+
+Import it in your code:
+```bash
+import "github.com/gomooth/httpsqs"
+
+```
+
+## Basic Examples
+
+```golang
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/gomooth/httpsqs"
+)
+
+func main() {
+	client := httpsqs.NewClient(&httpsqs.Config{
+		Addr:     "127.0.0.1:1218",
+		Timeout:  0,
+		Password: "",
+	})
+	name := "queue_test"
+	ctx := context.Background()
+	bs, _ := json.Marshal(struct {
+		Name  string
+		Title string
+		Age   int
+		Money int
+	}{
+		"张三", "无产阶级", 30, 10000,
+	})
+	pos, err := client.Put(ctx, name, string(bs))
+	fmt.Println(pos, err)
+}
+
+```

client.go

@@ -0,0 +1,214 @@
+package httpsqs
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/save95/xerror"
+)
+
+type client struct {
+	config *Config
+}
+
+func NewClient(config *Config) IClient {
+	return &client{config: config}
+}
+
+func (c *client) Put(ctx context.Context, name, data string) (int64, error) {
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "put")
+	values.Set("data", data)
+
+	body, headers, err := c.httpGet(ctx, values, []string{"pos"})
+	if err != nil {
+		return 0, xerror.Wrap(err, "httpsqs put failed")
+	}
+
+	switch body {
+	case "HTTPSQS_PUT_OK":
+		pos, _ := strconv.Atoi(headers["pos"])
+		return int64(pos), nil
+	case "HTTPSQS_PUT_ERROR":
+		return 0, xerror.New("httpsqs put err")
+	case "HTTPSQS_PUT_END":
+		return -1, xerror.New("httpsqs is full")
+	default:
+		return 0, xerror.Errorf("httpsqs put failed: %s", body)
+	}
+}
+
+func (c *client) Get(ctx context.Context, name string) (string, int64, error) {
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "get")
+
+	body, headers, err := c.httpGet(ctx, values, []string{"pos"})
+	if err != nil {
+		return "", 0, xerror.Wrap(err, "httpsqs get failed")
+	}
+
+	switch body {
+	case "HTTPSQS_GET_END": // 没有未取出的消息队列
+		return "", -1, nil
+	default:
+		pos, _ := strconv.Atoi(headers["pos"])
+		return body, int64(pos), nil
+	}
+}
+
+func (c *client) Status(ctx context.Context, name string) (*Status, error) {
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "status_json")
+
+	body, _, err := c.httpGet(ctx, values, nil)
+	if err != nil {
+		return nil, xerror.Wrap(err, "httpsqs status failed")
+	}
+
+	var res Status
+	if err := json.Unmarshal([]byte(body), &res); nil != err {
+		return nil, xerror.Wrapf(err, "httpsqs status failed: response not json: %s", body)
+	}
+
+	return &res, nil
+}
+
+func (c *client) View(ctx context.Context, name string, pos int64) (string, error) {
+	if pos <= 0 || pos > 1000000000 {
+		return "", xerror.New("input pos error")
+	}
+
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "view")
+	values.Set("pos", strconv.Itoa(int(pos)))
+
+	body, _, err := c.httpGet(ctx, values, nil)
+	if err != nil {
+		return "", xerror.Wrap(err, "httpsqs view failed")
+	}
+
+	return body, nil
+}
+
+func (c *client) Reset(ctx context.Context, name string) error {
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "reset")
+
+	body, _, err := c.httpGet(ctx, values, nil)
+	if err != nil {
+		return xerror.Wrap(err, "httpsqs reset failed")
+	}
+
+	switch body {
+	case "HTTPSQS_RESET_OK":
+		return nil
+	default:
+		return xerror.Errorf("httpsqs reset failed: %s", body)
+	}
+}
+
+func (c *client) SetMaxQueue(ctx context.Context, name string, max int) error {
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "maxqueue")
+	values.Set("num", strconv.Itoa(max))
+
+	body, _, err := c.httpGet(ctx, values, nil)
+	if err != nil {
+		return xerror.Wrap(err, "httpsqs set max_queue failed")
+	}
+
+	switch body {
+	case "HTTPSQS_MAXQUEUE_OK":
+		return nil
+	case "HTTPSQS_MAXQUEUE_CANCEL":
+		return xerror.New("httpsqs option cancel")
+	default:
+		return xerror.Errorf("httpsqs set max_queue failed: %s", body)
+	}
+}
+
+func (c *client) SetSyncTime(ctx context.Context, name string, duration time.Duration) error {
+	num := int(duration.Seconds())
+	if num < 1 || num > 1000000000 {
+		return xerror.New("input num error")
+	}
+
+	values := url.Values{}
+	values.Set("name", name)
+	values.Set("opt", "synctime")
+	values.Set("num", strconv.Itoa(num))
+
+	body, _, err := c.httpGet(ctx, values, nil)
+	if err != nil {
+		return xerror.Wrap(err, "httpsqs set sync_time failed")
+	}
+
+	switch body {
+	case "HTTPSQS_SYNCTIME_OK":
+		return nil
+	case "HTTPSQS_SYNCTIME_CANCEL":
+		return xerror.New("httpsqs option cancel")
+	default:
+		return xerror.Errorf("httpsqs set synctime failed: %s", body)
+	}
+}
+
+func (c *client) httpGet(ctx context.Context, values url.Values, parseHeaders []string) (string, map[string]string, error) {
+	furl := fmt.Sprintf("http://%s/", c.config.Addr)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, furl, nil)
+	if nil != err {
+		return "", nil, err
+	}
+
+	if len(c.config.Password) > 0 {
+		values.Set("auth", c.config.Password)
+	}
+	req.URL.RawQuery = values.Encode()
+
+	httpClient := http.DefaultClient
+	if c.config.Timeout > 0 {
+		httpClient.Timeout = c.config.Timeout
+	} else {
+		httpClient.Timeout = 5 * time.Second
+	}
+	resp, err := httpClient.Do(req)
+	if nil != err {
+		return "", nil, err
+	}
+
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+
+	bodyBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", nil, err
+	}
+
+	body := strings.TrimSpace(string(bodyBytes))
+	switch body {
+	case "HTTPSQS_ERROR": // 全局错误
+		return "", nil, xerror.New("httpsqs global error")
+	case "HTTPSQS_AUTH_FAILED": // 密码校验失败
+		return "", nil, xerror.New("httpsqs auth error")
+	default:
+		headerResult := make(map[string]string)
+		for _, header := range parseHeaders {
+			headerResult[header] = strings.TrimSpace(resp.Header.Get(header))
+		}
+		return body, headerResult, nil
+	}
+}