delay

README

delay-queue

delayqueue client for delayqueue

Client

client = delay.NewClient("delay_host")

Publish Message

req := &delay.Req{
    Topic:     "order",
    Id:        "orderId"
    Delay:     20*60,                     // 20minute
    TimeToRun: 60,                        // ttr:60s
    Body:      "", 
}
client.Push(req)

Consume Message

import ("github.com/youngbloood/delay")

func cycle(){
    pool := make(chan *delay.PopResp, 1000)

    // 轮巡获取消息
    go client.RoundPop("order", pool)

    // 处理消息的handler
    client.HandleMap["order"] = func(pr *delay.PopResp) error {
        // handle pr
        // ...
        return nil
    }

    // handle with sync
    go client.HandleSync("order", pool)
    // handle with async
    go client.HandleAsync("order", pool)

}

Documentation

Index

• type Client
  • func NewClient(host string) *Client
  • func (d *Client) Delete(id string) (*Req, error)
  • func (d *Client) Finish(id string) error
  • func (d *Client) Get(id string) (*Req, error)
  • func (d *Client) HandleAsync(topic string, queue <-chan *PopResp)
  • func (d *Client) HandleSync(topic string, queue <-chan *PopResp) error
  • func (d *Client) Pop(topic string) (pr *PopResp, err error)
  • func (d *Client) Push(req *Req) (*Resp, error)
  • func (d *Client) RoundPop(topic string, queue chan<- *PopResp) error
  • func (d *Client) SetClient(client *http.Client)
• type PopResp
• type Req
  • func (r *Req) Marshal() ([]byte, error)
  • func (r *Req) Read() io.Reader
• type Resp
  • func (r *Resp) GetError() error
  • func (r *Resp) HasError() bool
  • func (r *Resp) Unmarshal(bts []byte) error

Types

type Client

type Client struct {

	// topic:handle
	HandleMap map[string]func(*PopResp) error
	// contains filtered or unexported fields
}

func NewClient

func NewClient(host string) *Client

func (*Client) Delete

func (d *Client) Delete(id string) (*Req, error)

func (*Client) Finish

func (d *Client) Finish(id string) error

func (*Client) Get

func (d *Client) Get(id string) (*Req, error)

func (*Client) HandleAsync

func (d *Client) HandleAsync(topic string, queue <-chan *PopResp)

func (*Client) HandleSync

func (d *Client) HandleSync(topic string, queue <-chan *PopResp) error

func (*Client) Pop

func (d *Client) Pop(topic string) (pr *PopResp, err error)

func (*Client) Push

func (d *Client) Push(req *Req) (*Resp, error)

func (*Client) RoundPop

func (d *Client) RoundPop(topic string, queue chan<- *PopResp) error

func (*Client) SetClient

func (d *Client) SetClient(client *http.Client)

type PopResp

type PopResp struct {
	Id   string `json:"id"`
	Body string `json:"body"`
}

type Req

type Req struct {
	Topic     string `json:"topic,omitempty"` // Job类型
	Id        string `json:"id,omitempty"`    // Job唯一标识(确保job唯一)
	Delay     int64  `json:"delay,omitempty"` // Job需要延迟的时间, 单位：秒
	TimeToRun int64  `json:"ttr,omitempty"`   // Job执行超时时间, 单位：秒
	Body      string `json:"body,omitempty"`  // Job的内容，供消费者做具体的业务处理，如果是json格式需转义
}

func (*Req) Marshal

func (r *Req) Marshal() ([]byte, error)

func (*Req) Read

func (r *Req) Read() io.Reader

type Resp

type Resp struct {
	Code    int         `json:"code"`    //状态码	0: 成功 非0: 失败
	Message string      `json:"message"` // 状态描述信息
	Data    interface{} `json:"data"`    // 附加信息
}

func (*Resp) GetError

func (r *Resp) GetError() error

func (*Resp) HasError

func (r *Resp) HasError() bool

func (*Resp) Unmarshal

func (r *Resp) Unmarshal(bts []byte) error