SDKs
The SDKs for Node.js, Python and Go sign requests, try again when the service is briefly busy, and come with a ready-made worker that waits for messages, runs your code and deletes each message it finished. Install the one for your language:
| Language | Needs | Install |
|---|---|---|
| Node.js | 18.17+ | npm install @thailandhosting/queue-service |
| Python | 3.9+ | pip install thailandhosting-queue-service |
| Go | 1.23+ | go get github.com/mrzack99s/thailandhosting-queue-service-sdk-go |
Give the client the queue’s region and an access key from the Console (kept in environment variables, not in your code), and reuse one client in your program.
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})You can also create and change queues, and receive and delete messages yourself:
// 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, delayedA message can arrive more than once (for example when a worker stops before deleting it). Make your code safe to run twice, such as checking an order was not already charged.
The REST API
Each region’s endpoint is https://queues.thailandhosting.com/<region> and the API lives at …/<region>/v1/queues. Bodies and answers are JSON, and every duration is in seconds.
| Request | What it does |
|---|---|
GET /v1/queues | The queues the key can reach |
PUT /v1/queues/{name} | Create a queue or change its settings (the key must manage queues), e.g. {"visibilityTimeout":60,"maxReceives":5,"deadLetterQueue":"orders-dlq"} |
GET /v1/queues/{name} | Settings and message counts (waiting, in progress, delayed) |
POST /v1/queues/{name}/messages | Send one message {"body":"…","attributes":{…},"delay":0} or up to 10 {"messages":[…]} |
POST /v1/queues/{name}/receive | Receive up to 10 messages {"max":10,"wait":20,"visibilityTimeout":120}, each with id, body, receipt and receives |
POST /v1/queues/{name}/delete | Delete finished messages {"receipts":["…"]} (up to 10) |
POST /v1/queues/{name}/visibility | Take more time, or hand a message back at once {"receipt":"…","timeout":0} |
POST /v1/queues/{name}/purge | Remove every message in the queue |
DELETE /v1/queues/{name} | Delete the queue and its messages |
Signing requests
Each request is signed with HTTP Message Signatures (RFC 9421) using your secret (HMAC-SHA256), covering the method, host, path, query and the body’s Content-Digest (RFC 9530), so the secret is never sent. Your machine’s clock must be within 5 minutes of ours. Errors are Problem Details (RFC 9457) with a code.
Without an SDK: Node.js
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] });
}Without an SDK: Python
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"]]})A message (body and attributes) is at most 256 KiB; a send carries up to 10 messages and 1 MiB; a receive returns up to 10. A queue can have up to 120,000 messages received and not yet deleted. A Project can have up to 1,000 queues and 50 access keys.
If something goes wrong
401 “invalid_signature”
Check the key ID and secret, sign the full path including /<region>, and keep your clock right (turn on NTP).
409 “invalid_receipt” on delete
The message passed its visibility timeout and went to another worker, so the old receipt no longer works. Raise the visibility timeout and make the job safe to repeat.
429 “queue_full” or “too_many_in_flight”
The queue holds too many messages, or more than 120,000 are received and not deleted. Add workers and delete finished messages.
421 “wrong_region”
The queue is in another region. Use the endpoint given in the answer, or call without a region in the path.