Push sends a single message to the queue. The body is base64-encoded and sent with content_type=bytes so the consumer receives the raw flatbuffers payload. Returns an error on any non-2xx response or transport failure.
(ctx context.Context, payload []byte)
| 40 | // sent with content_type=bytes so the consumer receives the raw flatbuffers |
| 41 | // payload. Returns an error on any non-2xx response or transport failure. |
| 42 | func (c *Client) Push(ctx context.Context, payload []byte) error { |
| 43 | base := c.BaseURL |
| 44 | if base == "" { |
| 45 | base = defaultBaseURL |
| 46 | } |
| 47 | url := fmt.Sprintf( |
| 48 | "%s/client/v4/accounts/%s/queues/%s/messages", |
| 49 | base, c.AccountID, c.QueueID, |
| 50 | ) |
| 51 | |
| 52 | // content_type=text with a base64 string body. The Cloudflare HTTP API |
| 53 | // rejects content_type=bytes; the backend consumer base64-decodes string |
| 54 | // bodies before handing them to the flatbuffers decoder. |
| 55 | jsonBody, err := json.Marshal(pushBody{ |
| 56 | Body: base64.StdEncoding.EncodeToString(payload), |
| 57 | ContentType: "text", |
| 58 | }) |
| 59 | if err != nil { |
| 60 | return fmt.Errorf("marshal push body: %w", err) |
| 61 | } |
| 62 | |
| 63 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(jsonBody)) |
| 64 | if err != nil { |
| 65 | return fmt.Errorf("build request: %w", err) |
| 66 | } |
| 67 | req.Header.Set("Authorization", "Bearer "+c.Token) |
| 68 | req.Header.Set("Content-Type", "application/json") |
| 69 | |
| 70 | resp, err := c.HTTP.Do(req) |
| 71 | if err != nil { |
| 72 | return fmt.Errorf("cloudflare push: %w", err) |
| 73 | } |
| 74 | defer resp.Body.Close() |
| 75 | |
| 76 | if resp.StatusCode/100 != 2 { |
| 77 | snippet, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) |
| 78 | return fmt.Errorf("cloudflare push: status %d: %s", resp.StatusCode, string(snippet)) |
| 79 | } |
| 80 | return nil |
| 81 | } |