Retries and dead letters: a webhook sender

const stop = new AbortController()
process.once('SIGTERM', () => stop.abort())       // a deploy stops the sender cleanly

await queen.queue('webhooks')
  .group('sender')
  .subscriptionMode('all')
  .each()
  .consume(async (msg) => {
    const res = await fetch(msg.data.url, {
      method: 'POST',
      headers: { 'content-type': 'application/json', 'idempotency-key': msg.transactionId },
      body: JSON.stringify(msg.data.event),
    })
    if (!res.ok) throw new Error(`HTTP ${res.status}`)  // retried, then dead-lettered
  }, { signal: stop.signal })

import httpx

stop = asyncio.Event()
loop = asyncio.get_running_loop()
loop.add_signal_handler(signal.SIGTERM, stop.set)     # a deploy stops the sender cleanly

async with httpx.AsyncClient() as client:
    async def call(msg):
        res = await client.post(
            msg["data"]["url"],
            json=msg["data"]["event"],
            headers={"idempotency-key": msg["transactionId"]},
        )
        res.raise_for_status()                        # retried, then dead-lettered

    await (
        queen.queue("webhooks")
        .group("sender")
        .subscription_mode("all")
        .each()
        .consume(call, signal=stop)
    )

ctx, stop := signal.NotifyContext(ctx, syscall.SIGTERM) // a deploy stops it cleanly
defer stop()

err := client.Queue("webhooks").
	Group("sender").
	SubscriptionMode(queen.SubscriptionModeAll).
	Each().
	Consume(ctx, func(ctx context.Context, msg *queen.Message) error {
		event, err := json.Marshal(msg.Data["event"])
		if err != nil {
			return err
		}
		req, err := http.NewRequestWithContext(ctx, http.MethodPost,
			msg.Data["url"].(string), bytes.NewReader(event))
		if err != nil {
			return err
		}
		req.Header.Set("Content-Type", "application/json")
		req.Header.Set("Idempotency-Key", msg.TransactionID)
		res, err := http.DefaultClient.Do(req)
		if err != nil {
			return err
		}
		res.Body.Close()
		if res.StatusCode >= 300 {
			return fmt.Errorf("HTTP %d", res.StatusCode) // retried, then dead-lettered
		}
		return nil
	}).
	Execute(ctx) // returns once ctx is cancelled
if err != nil {
	log.Fatal(err)
}

use queen_mq::Cancel;
use tokio::signal::unix::{signal, SignalKind};

let stop = Cancel::new();
let on_sigterm = stop.clone();
tokio::spawn(async move {                   // a deploy stops the sender cleanly
    if let Ok(mut sigterm) = signal(SignalKind::terminate()) {
        sigterm.recv().await;
        on_sigterm.cancel();
    }
});

let http = reqwest::Client::new();
queen
    .queue("webhooks")
    .group("sender")
    .subscription_mode(SubscriptionMode::All)
    .poll_timeout(Duration::from_secs(5))   // a stop waits for the poll in flight (5 s)
    .cancel(stop)
    .consume(move |msg: Message| {
        let http = http.clone();
        async move {
            let res = http
                .post(msg.data["url"].as_str().unwrap_or_default())
                .header("idempotency-key", &msg.transaction_id)
                .json(&msg.data["event"])
                .send()
                .await
                .map_err(|e| e.to_string())?;
            if !res.status().is_success() {
                // an Err is retried after retryDelay, then dead-lettered
                return Err(format!("HTTP {}", res.status().as_u16()));
            }
            Ok(())
        }
    })
    .await?;

// A deploy's SIGTERM stops the sender cleanly: consume() handles it itself.
$http = new GuzzleHttp\Client();

$queen->queue('webhooks')
    ->group('sender')
    ->subscriptionMode('all')
    ->each()
    ->consume(function (array $msg) use ($http) {
        $res = $http->post($msg['data']['url'], [
            'headers' => ['idempotency-key' => $msg['transactionId']],
            'json' => $msg['data']['event'],
            'http_errors' => false,
        ]);
        if ($res->getStatusCode() >= 300) {         // retried, then dead-lettered
            throw new RuntimeException("HTTP {$res->getStatusCode()}");
        }
    })
    ->execute();

static std::atomic<bool> stop{false};            // static: a signal handler sees it
std::signal(SIGTERM, [](int) { stop = true; });  // a deploy stops the sender cleanly

client.queue("webhooks")
    .group("sender")
    .subscription_mode("all")
    .each()
    .wait(false)  // short polls: the stop flag is read between them
    .consume([](const json& msg) {
        const std::string url = msg["data"]["url"];       // http://host:port/path
        const size_t path = url.find('/', url.find("//") + 2);
        httplib::Client http(url.substr(0, path));
        httplib::Headers headers = {
            {"idempotency-key", msg["transactionId"].get<std::string>()}};
        auto res = http.Post(url.substr(path), headers, msg["data"]["event"].dump(),
                             "application/json");
        if (!res || res->status / 100 != 2) {             // retried, then dead-lettered
            throw std::runtime_error("HTTP " + std::to_string(res ? res->status : 0));
        }
    }, &stop);