Method queueUpdate

client.go:2873–2912

(ctx context.Context, executorTx riverdriver.ExecutorTx, name string, params *QueueUpdateParams)

}

func (c *Client[TTx]) queueUpdate(ctx context.Context, executorTx riverdriver.ExecutorTx, name string, params *QueueUpdateParams) (*rivertype.Queue, *controlEventPayload, error) {
	updateMetadata := len(params.Metadata) > 0

	queue, err := executorTx.QueueUpdate(ctx, &riverdriver.QueueUpdateParams{
		Metadata:         params.Metadata,
		MetadataDoUpdate: updateMetadata,
		Name:             name,
		Schema:           c.config.Schema,
	})
	if err != nil {
		return nil, nil, err
	}

	if !updateMetadata {
		return queue, nil, err
	}

	controlEvent := &controlEventPayload{
		Action:   controlActionMetadataChanged,
		Metadata: params.Metadata,
		Queue:    queue.Name,
	}

	payload, err := json.Marshal(controlEvent)
	if err != nil {
		return nil, nil, err
	}

	if c.driver.SupportsListenNotify() {
		if err := executorTx.NotifyMany(ctx, &riverdriver.NotifyManyParams{
			Payload: []string{string(payload)},
			Schema:  c.config.Schema,
			Topic:   string(notifier.NotificationTopicControl),
		}); err != nil {
			return nil, nil, err
		}
	}

	return queue, controlEvent, nil
}

// Schema returns the configured schema for the client.
func (c *Client[TTx]) Schema() string { return c.config.Schema }