garage-k2v-go/poll_batch.go

84 lines
1.9 KiB
Go

package k2v
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"time"
)
type PollRangeQuery struct {
// Prefix restricts items to poll to those whose sort keys start with this prefix.
Prefix string `json:"prefix,omitempty"`
// Start is the sort key of the first item to poll.
Start string `json:"start,omitempty"`
// End is the sort key of the last item to poll (excluded).
End string `json:"end,omitempty"`
// SeenMarker is an opaque string returned by a previous PollRange call, that represents items already seen.
SeenMarker string `json:"seenMarker,omitempty"`
}
type PollRangeResponse struct {
SeenMarker string `json:"seenMarker"`
Items []SearchResultItem `json:"items"`
}
func (c *Client) PollRange(ctx context.Context, b Bucket, pk string, q PollRangeQuery, timeout time.Duration) (*PollRangeResponse, error) {
u, err := url.Parse(c.endpoint)
if err != nil {
return nil, err
}
u.Path = string(b) + "/" + pk
query := make(url.Values)
query.Set("poll_range", "")
u.RawQuery = query.Encode()
reqBody, err := json.Marshal(struct {
PollRangeQuery
Timeout int `json:"timeout,omitempty"`
}{
PollRangeQuery: q,
Timeout: int(timeout.Seconds()),
})
if err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, "SEARCH", u.String(), bytes.NewReader(reqBody))
if err != nil {
return nil, err
}
resp, err := c.executeRequest(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
switch resp.StatusCode {
case http.StatusOK:
break
case http.StatusNotModified:
return nil, NotModifiedTimeoutErr
default:
return nil, fmt.Errorf("http status code %d: %s", resp.StatusCode, body)
}
var result PollRangeResponse
if err := json.Unmarshal(body, &result); err != nil {
return nil, err
}
return &result, nil
}