97 lines
3.1 KiB
Go
97 lines
3.1 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"net/url"
|
||
"strconv"
|
||
)
|
||
|
||
// CreateSubscription 保留原幂等键和请求内容重报；SDK不会自动生成新键或重新指定起点。
|
||
func (c *Client) CreateSubscription(ctx context.Context, key string, in CreateSubscriptionRequest) (SubscriptionResult, error) {
|
||
var out SubscriptionResult
|
||
if !validID(in.QueueID) {
|
||
return out, errors.New("invalid queue id")
|
||
}
|
||
err := c.call(ctx, "POST", "/v1/subscriptions", key, in, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) GetSubscription(ctx context.Context, id string) (Subscription, error) {
|
||
var out Subscription
|
||
if !validID(id) {
|
||
return out, errors.New("invalid subscription id")
|
||
}
|
||
err := c.call(ctx, "GET", "/v1/subscriptions/"+id, "", nil, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) ConfigureSubscription(ctx context.Context, id, key string, in ConfigureSubscriptionRequest) (SubscriptionResult, error) {
|
||
var out SubscriptionResult
|
||
if !validID(id) {
|
||
return out, errors.New("invalid subscription id")
|
||
}
|
||
err := c.call(ctx, "POST", "/v1/subscriptions/"+id+":configure", key, in, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) ListSubscriptions(ctx context.Context, queue string, in SubscriptionPageRequest) (SubscriptionPage, error) {
|
||
var out SubscriptionPage
|
||
if !validID(queue) {
|
||
return out, errors.New("invalid queue id")
|
||
}
|
||
q, err := subscriptionQuery(in.Limit, in.Cursor)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
q.Set("queue_id", queue)
|
||
err = c.call(ctx, "GET", "/v1/subscriptions?"+q.Encode(), "", nil, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) ListEventDeliveries(ctx context.Context, id string, in DeliveryPageRequest) (DeliveryPage, error) {
|
||
var out DeliveryPage
|
||
if !validID(id) {
|
||
return out, errors.New("invalid subscription id")
|
||
}
|
||
q, err := subscriptionQuery(in.Limit, in.Cursor)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
switch in.State {
|
||
case "", DeliveryPending, DeliveryLeased, DeliveryAcked, DeliveryExpired:
|
||
default:
|
||
return out, errors.New("invalid delivery state")
|
||
}
|
||
if in.State != "" {
|
||
q.Set("state", string(in.State))
|
||
}
|
||
err = c.call(ctx, "GET", "/v1/subscriptions/"+id+"/deliveries?"+q.Encode(), "", nil, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) GetEventDelivery(ctx context.Context, sub, id string) (DeliveryDetail, error) {
|
||
var out DeliveryDetail
|
||
if !validID(sub) || !validID(id) {
|
||
return out, errors.New("invalid subscription or delivery id")
|
||
}
|
||
err := c.call(ctx, "GET", "/v1/subscriptions/"+sub+"/deliveries/"+id, "", nil, &out)
|
||
return out, err
|
||
}
|
||
func (c *Client) RetryEventDelivery(ctx context.Context, sub, id, key string, in RetryDeliveryRequest) (RetryDeliveryResult, error) {
|
||
var out RetryDeliveryResult
|
||
if !validID(sub) || !validID(id) {
|
||
return out, errors.New("invalid subscription or delivery id")
|
||
}
|
||
err := c.call(ctx, "POST", "/v1/subscriptions/"+sub+"/deliveries/"+id+":retry", key, in, &out)
|
||
return out, err
|
||
}
|
||
func subscriptionQuery(limit int, cursor string) (url.Values, error) {
|
||
q := url.Values{}
|
||
if limit < 0 || limit > 100 || len(cursor) > 2048 {
|
||
return nil, errors.New("invalid subscription page")
|
||
}
|
||
if limit != 0 {
|
||
q.Set("limit", strconv.Itoa(limit))
|
||
}
|
||
if cursor != "" {
|
||
q.Set("cursor", cursor)
|
||
}
|
||
return q, nil
|
||
}
|