ใช้ SDK และ API

import { Client } from '@thailandhosting/queue-service';

const client = new Client({
  region: '<region>',
  accessKeyId: process.env.TH_QUEUE_ACCESS_KEY_ID,
  secretAccessKey: process.env.TH_QUEUE_SECRET_ACCESS_KEY,
});
const orders = client.queue('orders');

// Send
await orders.sendJSON({ orderId: 42 });
await orders.send('plain text', { attributes: { kind: 'email' }, delay: 30 });

// A worker: waits for messages, runs the handler, deletes what it finished
const stop = new AbortController();
process.on('SIGTERM', () => stop.abort());
await orders.process(async (m) => {
  const { orderId } = m.json();
  await ship(orderId); // throwing leaves the message to be tried again
}, { concurrency: 8, retryDelay: 30, signal: stop.signal });

import os
from thailandhosting_queueservice import Client

client = Client(
    access_key_id=os.environ["TH_QUEUE_ACCESS_KEY_ID"],
    secret_access_key=os.environ["TH_QUEUE_SECRET_ACCESS_KEY"],
    region="<region>",
)
orders = client.queue("orders")

# Send
orders.send_json({"order_id": 42})
orders.send("plain text", attributes={"kind": "email"}, delay=30)

# A worker: waits for messages, runs the handler, deletes what it finished
def handle(m):
    ship(m.json()["order_id"])  # raising leaves the message to be tried again

orders.process(handle, concurrency=8, retry_delay=30)

client, err := queueservice.New(queueservice.Config{
	Region:          "<region>",
	AccessKeyID:     os.Getenv("TH_QUEUE_ACCESS_KEY_ID"),
	SecretAccessKey: os.Getenv("TH_QUEUE_SECRET_ACCESS_KEY"),
})
if err != nil {
	log.Fatal(err)
}
orders := client.Queue("orders")

// Send
_, err = orders.SendJSON(ctx, map[string]any{"orderId": 42}, nil)

// A worker: waits for messages, runs the handler, deletes what it finished.
// It runs until ctx ends.
err = orders.Process(ctx, func(ctx context.Context, m queueservice.Message) error {
	var order struct{ OrderID int }
	if err := m.JSON(&order); err != nil {
		return err
	}
	return ship(ctx, order.OrderID) // an error leaves the message to be tried again
}, &queueservice.ProcessOptions{Concurrency: 8, RetryDelay: 30 * time.Second})

// Create a queue or change its settings (the key must manage queues)
await client.putQueue('orders', { visibilityTimeout: 120, maxReceives: 5, deadLetterQueue: 'orders-dlq' });

// Receive by hand
for (const m of await orders.receive({ max: 10, wait: 20 })) {
  await handle(m.body);
  await orders.delete(m.receipt);
}

const info = await client.getQueue('orders'); // info.messages: visible, inFlight, delayed

import crypto from 'node:crypto';
const KEY_ID = 'YOUR_KEY_ID', SECRET = 'YOUR_SECRET';
const ENDPOINT = 'https://queues.thailandhosting.com/<region>';

async function call(method, path, body) {
  const url = new URL(ENDPOINT + path);
  const data = body ? JSON.stringify(body) : '';
  const headers = { 'Content-Type': 'application/json' };
  let covered = '"@method" "@authority" "@path" "@query"';
  let base = `"@method": ${method}\n"@authority": ${url.host}\n"@path": ${url.pathname}\n"@query": ?${url.search.slice(1)}\n`;
  if (data) {
    headers['Content-Digest'] = `sha-256=:${crypto.createHash('sha256').update(data).digest('base64')}:`;
    covered += ' "content-digest"';
    base += `"content-digest": ${headers['Content-Digest']}\n`;
  }
  const params = `(${covered});created=${Math.floor(Date.now() / 1000)};keyid="${KEY_ID}";alg="hmac-sha256"`;
  base += `"@signature-params": ${params}`;
  headers['Signature-Input'] = `th=${params}`;
  headers['Signature'] = `th=:${crypto.createHmac('sha256', SECRET).update(base).digest('base64')}:`;
  const res = await fetch(url, { method, headers, body: data || undefined });
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  return res.status === 204 ? null : res.json();
}

await call('POST', '/v1/queues/orders/messages', { body: JSON.stringify({ orderId: 42 }) });
const { messages } = await call('POST', '/v1/queues/orders/receive', { max: 10, wait: 20 });
for (const m of messages) {
  // ... do the work ...
  await call('POST', '/v1/queues/orders/delete', { receipts: [m.receipt] });
}

import base64, hashlib, hmac, json, time, urllib.request, urllib.parse

KEY_ID, SECRET = "YOUR_KEY_ID", "YOUR_SECRET"
ENDPOINT = "https://queues.thailandhosting.com/<region>"

def call(method, path, body=None):
    url = urllib.parse.urlsplit(ENDPOINT + path)
    data = json.dumps(body).encode() if body is not None else b""
    headers = {"Content-Type": "application/json"}
    covered = '"@method" "@authority" "@path" "@query"'
    base = f'"@method": {method}\n"@authority": {url.netloc}\n"@path": {url.path}\n"@query": ?{url.query}\n'
    if data:
        headers["Content-Digest"] = "sha-256=:" + base64.b64encode(hashlib.sha256(data).digest()).decode() + ":"
        covered += ' "content-digest"'
        base += f'"content-digest": {headers["Content-Digest"]}\n'
    params = f'({covered});created={int(time.time())};keyid="{KEY_ID}";alg="hmac-sha256"'
    base += f'"@signature-params": {params}'
    headers["Signature-Input"] = "th=" + params
    headers["Signature"] = "th=:" + base64.b64encode(hmac.new(SECRET.encode(), base.encode(), hashlib.sha256).digest()).decode() + ":"
    req = urllib.request.Request(url.geturl(), data=data or None, method=method, headers=headers)
    with urllib.request.urlopen(req) as res:
        return json.loads(res.read() or "null")

call("POST", "/v1/queues/orders/messages", {"body": json.dumps({"orderId": 42})})
for m in call("POST", "/v1/queues/orders/receive", {"max": 10, "wait": 20})["messages"]:
    # ... do the work ...
    call("POST", "/v1/queues/orders/delete", {"receipts": [m["receipt"]]})