# Messages ## Push Message `client.Queues.Messages.Push(ctx, queueID, params) (*MessagePushResponse, error)` **post** `/accounts/{account_id}/queues/{queue_id}/messages` Push a message to a Queue ### Parameters - `queueID string` A Resource identifier. - `params MessagePushParams` - `AccountID param.Field[string]` Path param: A Resource identifier. - `Body param.Field[string]` Body param - `ContentType param.Field[MessagePushParamsMqQueueMessageTextContentType]` Body param - `const MessagePushParamsMqQueueMessageTextContentTypeText MessagePushParamsMqQueueMessageTextContentType = "text"` - `DelaySeconds param.Field[float64]` Body param: The number of seconds to wait for attempting to deliver this message to consumers ### Returns - `type MessagePushResponse struct{…}` - `Metadata MessagePushResponseMetadata` - `Metrics MessagePushResponseMetadataMetrics` Best-effort metrics for the queue. Values may be approximate due to the distributed nature of queues. - `BacklogBytes float64` The size in bytes of unacknowledged messages in the queue. - `BacklogCount float64` The number of unacknowledged messages in the queue. - `OldestMessageTimestampMs float64` Unix timestamp in milliseconds of the oldest unacknowledged message in the queue. Returns 0 if unknown. ### Example ```go package main import ( "context" "fmt" "github.com/cloudflare/cloudflare-go" "github.com/cloudflare/cloudflare-go/option" "github.com/cloudflare/cloudflare-go/queues" ) func main() { client := cloudflare.NewClient( option.WithAPIToken("Sn3lZJTBX6kkg7OdcBUAxOO963GEIyGQqnFTOFYY"), ) response, err := client.Queues.Messages.Push( context.TODO(), "023e105f4ecef8ad9ca31a8372d0c353", queues.MessagePushParams{ AccountID: cloudflare.F("023e105f4ecef8ad9ca31a8372d0c353"), }, ) if err != nil { panic(err.Error()) } fmt.Printf("%+v\n", response.Metadata) } ``` #### Response ```json { "errors": [ { "code": 7003, "message": "No route for the URI", "documentation_url": "documentation_url", "source": { "pointer": "pointer" } } ], "messages": [ "string" ], "result": { "metadata": { "metrics": { "backlog_bytes": 1024, "backlog_count": 5, "oldest_message_timestamp_ms": 1710950954154 } } }, "success": true } ``` ## Acknowledge + Retry Queue Messages `client.Queues.Messages.Ack(ctx, queueID, params) (*MessageAckResponse, error)` **post** `/accounts/{account_id}/queues/{queue_id}/messages/ack` Acknowledge + Retry messages from a Queue ### Parameters - `queueID string` A Resource identifier. - `params MessageAckParams` - `AccountID param.Field[string]` Path param: A Resource identifier. - `Acks param.Field[[]MessageAckParamsAck]` Body param - `LeaseID string` An ID that represents an "in-flight" message that has been pulled from a Queue. You must hold on to this ID and use it to acknowledge this message. - `Retries param.Field[[]MessageAckParamsRetry]` Body param - `DelaySeconds float64` The number of seconds to delay before making the message available for another attempt. - `LeaseID string` An ID that represents an "in-flight" message that has been pulled from a Queue. You must hold on to this ID and use it to acknowledge this message. ### Returns - `type MessageAckResponse struct{…}` - `AckCount float64` The number of messages that were succesfully acknowledged. - `RetryCount float64` The number of messages that were succesfully retried. - `Warnings map[string, string]` Map of lease IDs to warning messages encountered during acknowledgement. ### Example ```go package main import ( "context" "fmt" "github.com/cloudflare/cloudflare-go" "github.com/cloudflare/cloudflare-go/option" "github.com/cloudflare/cloudflare-go/queues" ) func main() { client := cloudflare.NewClient( option.WithAPIToken("Sn3lZJTBX6kkg7OdcBUAxOO963GEIyGQqnFTOFYY"), ) response, err := client.Queues.Messages.Ack( context.TODO(), "023e105f4ecef8ad9ca31a8372d0c353", queues.MessageAckParams{ AccountID: cloudflare.F("023e105f4ecef8ad9ca31a8372d0c353"), }, ) if err != nil { panic(err.Error()) } fmt.Printf("%+v\n", response.AckCount) } ``` #### Response ```json { "errors": [ { "code": 7003, "message": "No route for the URI", "documentation_url": "documentation_url", "source": { "pointer": "pointer" } } ], "messages": [ "string" ], "result": { "ackCount": 5, "retryCount": 5, "warnings": { "foo": "string" } }, "success": true } ``` ## Pull Queue Messages `client.Queues.Messages.Pull(ctx, queueID, params) (*MessagePullResponse, error)` **post** `/accounts/{account_id}/queues/{queue_id}/messages/pull` Pull a batch of messages from a Queue ### Parameters - `queueID string` A Resource identifier. - `params MessagePullParams` - `AccountID param.Field[string]` Path param: A Resource identifier. - `BatchSize param.Field[float64]` Body param: The maximum number of messages to include in a batch. - `VisibilityTimeoutMs param.Field[float64]` Body param: The number of milliseconds that a message is exclusively leased. After the timeout, the message becomes available for another attempt. ### Returns - `type MessagePullResponse struct{…}` - `MessageBacklogCount float64` The number of unacknowledged messages in the queue. - `Messages []MessagePullResponseMessage` - `ID string` - `Attempts float64` - `Body string` - `LeaseID string` An ID that represents an "in-flight" message that has been pulled from a Queue. You must hold on to this ID and use it to acknowledge this message. - `Metadata unknown` - `TimestampMs float64` - `Metadata MessagePullResponseMetadata` - `Metrics MessagePullResponseMetadataMetrics` Best-effort metrics for the queue. Values may be approximate due to the distributed nature of queues. - `BacklogBytes float64` The size in bytes of unacknowledged messages in the queue. - `BacklogCount float64` The number of unacknowledged messages in the queue. - `OldestMessageTimestampMs float64` Unix timestamp in milliseconds of the oldest unacknowledged message in the queue. Returns 0 if unknown. ### Example ```go package main import ( "context" "fmt" "github.com/cloudflare/cloudflare-go" "github.com/cloudflare/cloudflare-go/option" "github.com/cloudflare/cloudflare-go/queues" ) func main() { client := cloudflare.NewClient( option.WithAPIToken("Sn3lZJTBX6kkg7OdcBUAxOO963GEIyGQqnFTOFYY"), ) response, err := client.Queues.Messages.Pull( context.TODO(), "023e105f4ecef8ad9ca31a8372d0c353", queues.MessagePullParams{ AccountID: cloudflare.F("023e105f4ecef8ad9ca31a8372d0c353"), }, ) if err != nil { panic(err.Error()) } fmt.Printf("%+v\n", response.MessageBacklogCount) } ``` #### Response ```json { "errors": [ { "code": 7003, "message": "No route for the URI", "documentation_url": "documentation_url", "source": { "pointer": "pointer" } } ], "messages": [ "string" ], "result": { "message_backlog_count": 5, "messages": [ { "id": "b01b5594f784d0165c2985833f5660dd", "attempts": 1, "body": "hello world", "lease_id": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0..Q8p21d7dceR6vUfwftONdQ.JVqZgAS-Zk7MqmqccYtTHeeMElNHaOMigeWdb8LyMOg.T2_HV99CYzGaQuhTyW8RsgbnpTRZHRM6N7UoSaAKeK0", "metadata": { "CF-Content-Type": "text", "CF-sourceMessageSource": "dash" }, "timestamp_ms": 1710950954154 } ], "metadata": { "metrics": { "backlog_bytes": 1024, "backlog_count": 5, "oldest_message_timestamp_ms": 1710950954154 } } }, "success": true } ``` ## Push Message Batch `client.Queues.Messages.BulkPush(ctx, queueID, params) (*MessageBulkPushResponse, error)` **post** `/accounts/{account_id}/queues/{queue_id}/messages/batch` Push a batch of message to a Queue ### Parameters - `queueID string` A Resource identifier. - `params MessageBulkPushParams` - `AccountID param.Field[string]` Path param: A Resource identifier. - `DelaySeconds param.Field[float64]` Body param: The number of seconds to wait for attempting to deliver this batch to consumers - `Messages param.Field[[]MessageBulkPushParamsMessage]` Body param - `type MessageBulkPushParamsMessagesMqQueueMessageText struct{…}` - `Body string` - `ContentType MessageBulkPushParamsMessagesMqQueueMessageTextContentType` - `const MessageBulkPushParamsMessagesMqQueueMessageTextContentTypeText MessageBulkPushParamsMessagesMqQueueMessageTextContentType = "text"` - `DelaySeconds float64` The number of seconds to wait for attempting to deliver this message to consumers - `type MessageBulkPushParamsMessagesMqQueueMessageJson struct{…}` - `Body unknown` - `ContentType MessageBulkPushParamsMessagesMqQueueMessageJsonContentType` - `const MessageBulkPushParamsMessagesMqQueueMessageJsonContentTypeJson MessageBulkPushParamsMessagesMqQueueMessageJsonContentType = "json"` - `DelaySeconds float64` The number of seconds to wait for attempting to deliver this message to consumers ### Returns - `type MessageBulkPushResponse struct{…}` - `Metadata MessageBulkPushResponseMetadata` - `Metrics MessageBulkPushResponseMetadataMetrics` Best-effort metrics for the queue. Values may be approximate due to the distributed nature of queues. - `BacklogBytes float64` The size in bytes of unacknowledged messages in the queue. - `BacklogCount float64` The number of unacknowledged messages in the queue. - `OldestMessageTimestampMs float64` Unix timestamp in milliseconds of the oldest unacknowledged message in the queue. Returns 0 if unknown. ### Example ```go package main import ( "context" "fmt" "github.com/cloudflare/cloudflare-go" "github.com/cloudflare/cloudflare-go/option" "github.com/cloudflare/cloudflare-go/queues" ) func main() { client := cloudflare.NewClient( option.WithAPIToken("Sn3lZJTBX6kkg7OdcBUAxOO963GEIyGQqnFTOFYY"), ) response, err := client.Queues.Messages.BulkPush( context.TODO(), "023e105f4ecef8ad9ca31a8372d0c353", queues.MessageBulkPushParams{ AccountID: cloudflare.F("023e105f4ecef8ad9ca31a8372d0c353"), }, ) if err != nil { panic(err.Error()) } fmt.Printf("%+v\n", response.Metadata) } ``` #### Response ```json { "errors": [ { "code": 7003, "message": "No route for the URI", "documentation_url": "documentation_url", "source": { "pointer": "pointer" } } ], "messages": [ "string" ], "result": { "metadata": { "metrics": { "backlog_bytes": 1024, "backlog_count": 5, "oldest_message_timestamp_ms": 1710950954154 } } }, "success": true } ```