Skip to content

Update QueueConsumer to reflect the API #3023

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
76creates wants to merge 2 commits into from
Closed

Update QueueConsumer to reflect the API #3023

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1fd1108
fix: remove 'workers' from the queue consumer API paths
76creates Sep 4, 2024
49d0098
Merge branch 'cloudflare:master' into master
76creates Sep 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 16 additions & 13 deletions queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,20 +32,23 @@ type QueueProducer struct {
}

type QueueConsumer struct {
Name string `json:"-"`
Name string `json:"consumer_id,omitempty"`
Service string `json:"service,omitempty"`
ScriptName string `json:"script_name,omitempty"`
Environment string `json:"environment,omitempty"`
Settings QueueConsumerSettings `json:"settings,omitempty"`
QueueName string `json:"queue_name,omitempty"`
CreatedOn *time.Time `json:"created_on,omitempty"`
DeadLetterQueue string `json:"dead_letter_queue,omitempty"`
Type string `json:"type,omitempty"`
}

type QueueConsumerSettings struct {
BatchSize int `json:"batch_size,omitempty"`
MaxRetires int `json:"max_retries,omitempty"`
MaxWaitTime int `json:"max_wait_time_ms,omitempty"`
BatchSize int `json:"batch_size,omitempty"`
MaxRetires int `json:"max_retries,omitempty"`
MaxWaitTime int `json:"max_wait_time_ms,omitempty"`
RetryDelay int `json:"retry_delay,omitempty"`
VisibilityTimeout int `json:"visibility_timeout_ms,omitempty"`
}

type QueueListResponse struct {
Expand Down Expand Up @@ -125,7 +128,7 @@ func (api *API) ListQueues(ctx context.Context, rc *ResourceContainer, params Li
var qResponse QueueListResponse
for {
qResponse = QueueListResponse{}
uri := buildURI(fmt.Sprintf("/accounts/%s/workers/queues", rc.Identifier), params)
uri := buildURI(fmt.Sprintf("/accounts/%s/queues", rc.Identifier), params)

res, err := api.makeRequestContext(ctx, http.MethodGet, uri, nil)
if err != nil {
Expand Down Expand Up @@ -160,7 +163,7 @@ func (api *API) CreateQueue(ctx context.Context, rc *ResourceContainer, queue Cr
return Queue{}, ErrMissingQueueName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues", rc.Identifier)
uri := fmt.Sprintf("/accounts/%s/queues", rc.Identifier)
res, err := api.makeRequestContext(ctx, http.MethodPost, uri, queue)
if err != nil {
return Queue{}, fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand All @@ -185,7 +188,7 @@ func (api *API) DeleteQueue(ctx context.Context, rc *ResourceContainer, queueNam
return ErrMissingQueueName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s", rc.Identifier, queueName)
uri := fmt.Sprintf("/accounts/%s/queues/%s", rc.Identifier, queueName)
_, err := api.makeRequestContext(ctx, http.MethodDelete, uri, nil)
if err != nil {
return fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand All @@ -205,7 +208,7 @@ func (api *API) GetQueue(ctx context.Context, rc *ResourceContainer, queueName s
return Queue{}, ErrMissingQueueName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s", rc.Identifier, queueName)
uri := fmt.Sprintf("/accounts/%s/queues/%s", rc.Identifier, queueName)
res, err := api.makeRequestContext(ctx, http.MethodGet, uri, nil)
if err != nil {
return Queue{}, fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand All @@ -232,7 +235,7 @@ func (api *API) UpdateQueue(ctx context.Context, rc *ResourceContainer, params U
return Queue{}, ErrMissingQueueName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s", rc.Identifier, params.Name)
uri := fmt.Sprintf("/accounts/%s/queues/%s", rc.Identifier, params.Name)
res, err := api.makeRequestContext(ctx, http.MethodPut, uri, params)
if err != nil {
return Queue{}, fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand Down Expand Up @@ -273,7 +276,7 @@ func (api *API) ListQueueConsumers(ctx context.Context, rc *ResourceContainer, p
var qResponse ListQueueConsumersResponse
for {
qResponse = ListQueueConsumersResponse{}
uri := buildURI(fmt.Sprintf("/accounts/%s/workers/queues/%s/consumers", rc.Identifier, params.QueueName), params)
uri := buildURI(fmt.Sprintf("/accounts/%s/queues/%s/consumers", rc.Identifier, params.QueueName), params)

res, err := api.makeRequestContext(ctx, http.MethodGet, uri, nil)
if err != nil {
Expand Down Expand Up @@ -308,7 +311,7 @@ func (api *API) CreateQueueConsumer(ctx context.Context, rc *ResourceContainer,
return QueueConsumer{}, ErrMissingQueueName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s/consumers", rc.Identifier, params.QueueName)
uri := fmt.Sprintf("/accounts/%s/queues/%s/consumers", rc.Identifier, params.QueueName)
res, err := api.makeRequestContext(ctx, http.MethodPost, uri, params.Consumer)
if err != nil {
return QueueConsumer{}, fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand Down Expand Up @@ -338,7 +341,7 @@ func (api *API) DeleteQueueConsumer(ctx context.Context, rc *ResourceContainer,
return ErrMissingQueueConsumerName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s/consumers/%s", rc.Identifier, params.QueueName, params.ConsumerName)
uri := fmt.Sprintf("/accounts/%s/queues/%s/consumers/%s", rc.Identifier, params.QueueName, params.ConsumerName)
_, err := api.makeRequestContext(ctx, http.MethodDelete, uri, nil)
if err != nil {
return fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand All @@ -363,7 +366,7 @@ func (api *API) UpdateQueueConsumer(ctx context.Context, rc *ResourceContainer,
return QueueConsumer{}, ErrMissingQueueConsumerName
}

uri := fmt.Sprintf("/accounts/%s/workers/queues/%s/consumers/%s", rc.Identifier, params.QueueName, params.Consumer.Name)
uri := fmt.Sprintf("/accounts/%s/queues/%s/consumers/%s", rc.Identifier, params.QueueName, params.Consumer.Name)
res, err := api.makeRequestContext(ctx, http.MethodPut, uri, params.Consumer)
if err != nil {
return QueueConsumer{}, fmt.Errorf("%s: %w", errMakeRequestError, err)
Expand Down