client.go

package httpsqs

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/gomooth/xerror"
)

type client struct {
	config     *Config
	httpClient *http.Client
}

func NewClient(config *Config) IClient {
	timeout := config.Timeout
	if timeout <= 0 {
		timeout = 5 * time.Second
	}
	return &client{
		config: config,
		httpClient: &http.Client{
			Timeout: timeout,
			Transport: &http.Transport{
				MaxIdleConnsPerHost: 2,
				IdleConnTimeout:     90 * time.Second,
			},
		},
	}
}

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)

	if len(c.config.Password) > 0 {
		values.Set("auth", c.config.Password)
	}

	var lastErr error
	maxAttempts := c.config.MaxRetries + 1
	for attempt := 0; attempt < maxAttempts; attempt++ {
		if attempt > 0 {
			if err := c.waitBackoff(ctx, attempt); err != nil {
				return "", nil, err
			}
		}

		req, err := http.NewRequestWithContext(ctx, http.MethodGet, furl, nil)
		if err != nil {
			return "", nil, err
		}
		req.URL.RawQuery = values.Encode()

		resp, err := c.httpClient.Do(req)
		if err != nil {
			lastErr = err
			continue
		}

		return c.processResponse(resp, parseHeaders)
	}

	return "", nil, xerror.Wrap(lastErr, "httpsqs request failed after retries")
}

func (c *client) processResponse(resp *http.Response, parseHeaders []string) (string, map[string]string, error) {
	defer 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
	}
}

func (c *client) waitBackoff(ctx context.Context, attempt int) error {
	backoff := time.Duration(1<<uint(min(attempt-1, 24))) * 200 * time.Millisecond
	if backoff > 5*time.Second {
		backoff = 5 * time.Second
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(backoff):
		return nil
	}
}