tosync

README

将下游基于回调的异步能力，转为同步请求

时序图

示例代码

package tosync

import (
	"context"
)

func example() {
	ctx := context.Background()
	type SubmitReq struct {
		// 提交异步请求的入参
	}
	type SubmitResp struct{
		// 提交异步请求的返回值
	}
	type Resp struct {
		// 原来callback接口的入参
	}
	req := new(SubmitReq) // TODO 需要给req补充callbackURL的处理函数
	finalResp, err := ToSync[*SubmitReq, *Resp](ctx, req, func(ctx context.Context, req *SubmitReq) error {
		var resp SubmitResp
		resp, err := SubmitTask(ctx, req)
		if err != nil {
			return err
		}
		_ = resp // TODO 解析错误信息，转换为error
		return err
	})
	if err != nil {
		// TODO 处理错误
	}
	_ = finalResp // TODO 这个finalResp就是回调的结果
}

Documentation

Index

• Variables
• func CallbackHandler(ctx context.Context, r *http.Request) (err error)
• func Init(redisCli *redis.Client, cfg *Config) (err error)
• func ToSync[Req ReqI, CallbackData any](ctx context.Context, req Req, async func(context.Context, Req) error, ...) (data CallbackData, err error)
• type CallbackInfo
• type CallbackInfoParsed
• type Client
  • func (c *Client) CallbackHandler(ctx context.Context, r *http.Request) (err error)
  • func (c *Client) Regist(req ReqI) (*WaiterInfo, error)
  • func (c *Client) Release(info *WaiterInfo)
• type Config
  • func (c Config) Validate() error
• type Messager
• type Option
  • func (o *Option) SetClient(client *Client) *Option
  • func (o *Option) SetTimeout(timeout time.Duration) *Option
• type ReqI
• type WaiterInfo

Variables

var ErrInvalidSign = errors.New("invalid sign")

Functions

func CallbackHandler

func CallbackHandler(ctx context.Context, r *http.Request) (err error)

func Init

func Init(redisCli *redis.Client, cfg *Config) (err error)

func ToSync

func ToSync[Req ReqI, CallbackData any](ctx context.Context, req Req, async func(context.Context, Req) error, opts ...*Option) (data CallbackData, err error)

Types

type CallbackInfo

type CallbackInfo struct {
	AsyncID    string `json:"async_id"`
	Base64Body string `json:"base64_body"`
}

type CallbackInfoParsed

type CallbackInfoParsed struct {
	MsgID string // 用于消息队列的ack
	Body  []byte
}

type Client

type Client struct {
	// contains filtered or unexported fields
}

func (*Client) CallbackHandler

func (c *Client) CallbackHandler(ctx context.Context, r *http.Request) (err error)

func (*Client) Regist

func (c *Client) Regist(req ReqI) (*WaiterInfo, error)

func (*Client) Release

func (c *Client) Release(info *WaiterInfo)

type Config

type Config struct {
	CallbackURL      string `json:"callback_url" yaml:"callback_url" validate:"url"`              // 回调地址
	MaxCallbackBytes int64  `json:"max_callback_bytes" yaml:"max_callback_bytes" validate:"gt=0"` // 回调body限制
	Stream           string `json:"stream" yaml:"stream" validate:"gt=0"`                         // 回调stream key
	TimeoutSeconds   int    `json:"timeout_seconds" yaml:"timeout_seconds" validate:"gt=0"`       // 超时时间
}

func (Config) Validate

func (c Config) Validate() error

type Messager

type Messager interface {
	Pub(context.Context, []byte) (msgID string, err error)
	// DupSub: 需要让pub的msg被所有消费者都消费到，data: msgID -> msgData
	DupSub(context.Context) (data map[string][]byte, err error)
	Ack(ctx context.Context, msgID string) error
}

type Option

type Option struct {
	Client  *Client
	Timeout time.Duration
}

func (*Option) SetClient

func (o *Option) SetClient(client *Client) *Option

func (*Option) SetTimeout

func (o *Option) SetTimeout(timeout time.Duration) *Option

type ReqI

type ReqI interface {
	SetCallbackURL(url string)
	GetCallbackURL() string
}

type WaiterInfo

type WaiterInfo struct {
	AsyncID string
	State   string
	ResultC chan *CallbackInfoParsed
}