Produce and consume

//
// Queen MQ by example, 1 of 3: the smallest complete loop.
//
// Create a queue, push a handful of orders, consume them with a consumer group,
// acknowledge each one, and verify that every message arrived exactly once.
//
// Run it:
//   QUEEN_URL=http://localhost:6699 node 01-produce-consume.mjs
//
// The program checks its own outcome and exits non-zero if any check fails.

import { Queen } from 'queen-mq'

const QUEEN_URL = process.env.QUEEN_URL || 'http://localhost:6699'

// The name is prefixed per language so the JS, Python and Go examples can run
// against the same broker without colliding, and suffixed per run so a second
// run cannot inherit anything from the first.
const RUN_ID = Date.now().toString(36)
const QUEUE = `ex-js-produce-consume-${RUN_ID}`

// A consumer group is a named cursor over the queue. Every group sees every
// message; acking moves that group's cursor forward and affects no other group.
const GROUP = 'ex-js-shipping'

const ORDERS = [
  { orderId: 'A-1', customer: 'acme', total: 120.5 },
  { orderId: 'A-2', customer: 'acme', total: 12.0 },
  { orderId: 'B-1', customer: 'globex', total: 88.75 },
  { orderId: 'B-2', customer: 'globex', total: 4.2 },
  { orderId: 'C-1', customer: 'initech', total: 310.0 },
]

let checks = 0

function assert(condition, description) {
  if (!condition) throw new Error(description)
  checks++
  console.log(`  ok: ${description}`)
}

// handleSignals: false stops the client from installing its own SIGINT/SIGTERM
// handlers, because this script owns its shutdown through close() below.
const queen = new Queen({ url: QUEEN_URL, handleSignals: false })

try {
  console.log(`broker ${QUEEN_URL}`)

  // create() sends the whole configuration and is a full replace: keys you leave
  // out go back to the broker's defaults rather than keeping a previous value.
  await queen.queue(QUEUE).config({
    // How long a popped message stays invisible to other consumers before the
    // broker assumes the consumer died and hands it to someone else.
    leaseTime: 30,
    retryLimit: 3,
  }).create()
  console.log(`queue ${QUEUE} created`)

  console.log('\npushing')
  // push() accepts one item or an array. Each item's body is under `data`.
  // The client mints a UUIDv7 transactionId for any item that does not carry
  // one, and that id is the key the broker deduplicates on.
  const pushResults = await queen.queue(QUEUE).push(ORDERS.map(order => ({ data: order })))
  for (const result of pushResults) {
    console.log(`  ${result.transaction_id} -> ${result.status}`)
  }
  assert(pushResults.length === ORDERS.length, `broker answered for all ${ORDERS.length} items`)
  assert(pushResults.every(r => r.status === 'queued'), 'every item was accepted as queued')

  console.log('\nconsuming')
  const received = []
  await queen.queue(QUEUE)
    .group(GROUP)
    .each() // call the handler once per message instead of once per popped batch
    .batch(10) // up to 10 messages per poll
    .autoAck(false) // acknowledge by hand below, so the commit is visible in the code
    .limit(ORDERS.length) // stop the worker once it has handled this many messages
    .idleMillis(5000) // and stop anyway after 5s of silence, so a lost message fails the run instead of hanging it
    .consume(async (msg) => {
      console.log(`  ${msg.data.orderId} (${msg.data.customer}) from partition ${msg.partition}`)
      received.push(msg)

      // The acknowledgement is the commit. It moves this group's cursor past the
      // message and releases the lease; until it lands the message is only on
      // loan and would be redelivered when the lease expires.
      const ack = await queen.ack(msg, true, { group: GROUP })

      // A rejected ack still arrives as HTTP 200 with success: false on the item,
      // so the per-item flag is the only proof the broker took it.
      if (!ack.success) throw new Error(`ack rejected: ${ack.error}`)
    })

  console.log('\nchecking')
  assert(received.length === ORDERS.length, `consumed ${ORDERS.length} messages`)

  const arrivedIds = received.map(m => m.data.orderId)
  assert(new Set(arrivedIds).size === arrivedIds.length, 'no message was delivered twice')
  assert(
    ORDERS.every(order => arrivedIds.includes(order.orderId)),
    'every pushed order arrived at least once'
  )

  // The cursor for this group is now past every message, so a further poll on
  // the same group finds nothing. wait(false) makes it return immediately.
  const leftovers = await queen.queue(QUEUE).group(GROUP).batch(10).wait(false).pop()
  assert(leftovers.length === 0, `nothing is left for group ${GROUP}`)

  // Clean up. Only on success, so a failed run leaves the queue on the broker.
  await queen.queue(QUEUE).delete()

  console.log(`\nPASS: ${checks} checks`)
} catch (err) {
  console.error(`\nFAIL: ${err.message}`)
  process.exitCode = 1
} finally {
  // close() flushes buffers and destroys the HTTP dispatcher. Without it the
  // keep-alive sockets hold the Node event loop open and the process hangs.
  await queen.close()
}

"""Queen in Python: the smallest complete loop.

Create a queue, push a handful of orders, consume them with a consumer group,
acknowledge each one, and verify that every order arrived exactly once.

Run it with the client source on the path:

    cd /path/to/queen
    PYTHONPATH=clients/client-py QUEEN_URL=http://localhost:6699 \
        python3 examples/full/py/01_produce_consume.py

Exits 0 when every check passes, 1 on the first failure.
"""

import asyncio
import os
import sys
import uuid

from queen import Queen

BROKER_URL = os.environ.get("QUEEN_URL", "http://localhost:6699")

# A fresh queue name per run keeps this run independent of every previous one.
RUN_ID = uuid.uuid4().hex[:8]
QUEUE = f"ex-py-produce-consume-{RUN_ID}"

# A consumer group is a named cursor over the queue. The ack is what moves that
# cursor forward, so the group is the unit of "who has read how far".
GROUP = "receipts"

ORDERS = [
    {"orderId": "A-1", "customer": "acme", "amount": 120},
    {"orderId": "A-2", "customer": "acme", "amount": 40},
    {"orderId": "B-1", "customer": "globex", "amount": 250},
    {"orderId": "B-2", "customer": "globex", "amount": 15},
    {"orderId": "C-1", "customer": "initech", "amount": 999},
]


def check(condition, description):
    """Assert one claim and say so out loud.

    The example is documentation, so it must never print a success it did not
    verify. A failed check exits non-zero.
    """
    if not condition:
        print(f"FAIL  {description}")
        sys.exit(1)
    print(f"ok    {description}")


async def main():
    # The Python client is async throughout, and an async context manager, so
    # exiting the block flushes buffers and closes the HTTP client.
    async with Queen(BROKER_URL) as client:
        print(f"broker {BROKER_URL}")
        print(f"queue  {QUEUE}\n")

        # ------------------------------------------------------------------
        # 1. Create the queue.
        # ------------------------------------------------------------------
        # /api/v1/configure is a full replace: every key you leave out goes back
        # to its default, so send the whole configuration you want every time.
        created = await (
            client.queue(QUEUE)
            .config(
                {
                    "lease_time": 60,  # seconds a popped message stays claimed by its consumer
                    "retry_limit": 3,  # failed deliveries before the message lands in the DLQ
                    "dedup_window_seconds": 300,  # how far back the broker remembers transaction ids
                }
            )
            .create()
        )
        check(created.get("configured") is True, f"queue {QUEUE} created")
        check(created["options"]["leaseTime"] == 60, "lease time came back as configured")

        # ------------------------------------------------------------------
        # 2. Push the orders.
        # ------------------------------------------------------------------
        # One push call carries the whole list. The broker answers with one
        # result per item, in the order sent: queued, duplicate or failed.
        results = await client.queue(QUEUE).push([{"data": order} for order in ORDERS])

        for order, result in zip(ORDERS, results):
            print(f"      pushed {order['orderId']} -> {result['status']}")
        check(len(results) == len(ORDERS), f"broker answered for all {len(ORDERS)} pushed messages")
        check(
            all(r["status"] == "queued" for r in results),
            "every push was accepted as new work",
        )

        # ------------------------------------------------------------------
        # 3. Consume them.
        # ------------------------------------------------------------------
        received = []

        async def handle(message):
            # With batch(1) the Python handler receives a single message dict.
            # With batch(n) for n > 1 it receives a list instead, unless you add
            # .each() to unroll the batch one message at a time.
            order = message["data"]
            received.append(order)
            print(f"      consumed {order['orderId']} for {order['customer']}")
            # Returning without raising is the commit: the consumer acks the
            # message for this group, and that ack advances the group's cursor.
            # Raising here would nack it instead, and the broker would redeliver.

        await (
            client.queue(QUEUE)
            .group(GROUP)
            .batch(1)
            .limit(len(ORDERS))  # stop this worker after five messages so the example ends
            .idle_millis(5000)  # and stop anyway if the broker has nothing more to give
            .timeout_millis(2000)  # length of one long poll
            .consume(handle)
        )

        # ------------------------------------------------------------------
        # 4. Verify: everything, once.
        # ------------------------------------------------------------------
        pushed_ids = sorted(o["orderId"] for o in ORDERS)
        received_ids = sorted(o["orderId"] for o in received)

        check(len(received) == len(ORDERS), f"received {len(received)} messages, expected {len(ORDERS)}")
        check(len(set(received_ids)) == len(received_ids), "no order was delivered twice")
        check(received_ids == pushed_ids, "every pushed order came back")

        # A drained queue is the other half of "exactly once": the acks moved the
        # group's cursor past the last message, so a further pop finds nothing.
        leftover = await (
            client.queue(QUEUE).group(GROUP).batch(10).wait(True).timeout_millis(2000).pop()
        )
        check(leftover == [], f"nothing is left for group {GROUP}")

        # ------------------------------------------------------------------
        # 5. Clean up. Queues are cheap, but this one was for this run only.
        # ------------------------------------------------------------------
        deleted = await client.queue(QUEUE).delete()
        check(deleted.get("deleted") is True, f"queue {QUEUE} deleted")

        print("\nPASS 01_produce_consume")


asyncio.run(main())

// Produce and consume: the smallest complete Queen loop.
//
// Create a queue, push five messages, read them back through a consumer group,
// acknowledge them, and prove that every message arrived exactly once.
//
// Run it with:
//
//	QUEEN_URL=http://localhost:6699 go run ./produce-consume
package main

import (
	"context"
	"fmt"
	"os"
	"time"

	queen "github.com/smartpricing/queen/clients/client-go"
)

const (
	// A consumer group is a named cursor over the queue. Two groups reading the
	// same queue each receive every message once, independently of each other.
	consumerGroup = "ex-go-billing"

	messageCount = 5
)

// Queue names carry a per-language prefix so the JavaScript, Python and Go
// examples can run against the same broker without colliding, and a per-run
// suffix so every run starts from a queue that has never existed before: no
// messages, no consumer group cursor, no deduplication history.
var queueName = fmt.Sprintf("ex-go-produce-consume-%d", time.Now().UnixMilli())

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	if err := run(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "FAILED: %v\n", err)
		os.Exit(1)
	}
	fmt.Printf("OK: %d messages produced, consumed and acked exactly once\n", messageCount)
}

func run(ctx context.Context) error {
	brokerURL := os.Getenv("QUEEN_URL")
	if brokerURL == "" {
		brokerURL = "http://localhost:6699"
	}

	client, err := queen.New(brokerURL)
	if err != nil {
		return fmt.Errorf("create client: %w", err)
	}
	// The Go client installs no signal handlers, so shutting it down is your
	// job. Close flushes anything still held client side.
	defer client.Close(context.Background())

	fmt.Printf("broker: %s\n", brokerURL)

	// Creating a queue is a full replace of its configuration: every option you
	// leave out returns to the broker default, so send what you actually want.
	if _, err := client.Queue(queueName).Config(queen.QueueConfig{
		LeaseTime:  60, // seconds a popped message stays hidden from other consumers
		RetryLimit: 3,  // failed deliveries tolerated before the message is dead lettered
	}).Create().Execute(ctx); err != nil {
		return fmt.Errorf("create queue: %w", err)
	}
	fmt.Printf("queue %q ready\n", queueName)

	// Produce.
	//
	// pending maps every transactionId we pushed to whether we have seen it
	// come back. It is the ledger the assertions at the end are built on.
	pending := make(map[string]bool, messageCount)
	for i := 1; i <= messageCount; i++ {
		payload := map[string]interface{}{
			"orderId": fmt.Sprintf("order-%d", i),
			"amount":  10 * i,
		}

		responses, err := client.Queue(queueName).Push(payload).Execute(ctx)
		if err != nil {
			return fmt.Errorf("push %d: %w", i, err)
		}
		if len(responses) != 1 {
			return fmt.Errorf("push %d: expected 1 response, got %d", i, len(responses))
		}
		// The broker answers with one status per item, in request order.
		if responses[0].Status != "queued" {
			return fmt.Errorf("push %d: expected status queued, got %q (%s)",
				i, responses[0].Status, responses[0].Error)
		}

		// When you do not supply one, the client mints a UUIDv7 transactionId.
		// That id is the message identity: dedup and acknowledgement key on it.
		pending[responses[0].TransactionID] = false
		fmt.Printf("pushed order-%d as %s\n", i, responses[0].TransactionID)
	}

	// Consume.
	consumed := 0
	deadline := time.Now().Add(30 * time.Second)
	for consumed < messageCount && time.Now().Before(deadline) {
		// Pop does not long poll by default in Go. Wait(true) is what makes the
		// broker hold the request open until something is available.
		msgs, err := client.Queue(queueName).
			Group(consumerGroup).
			Batch(messageCount).
			Wait(true).
			TimeoutMillis(5000).
			Pop(ctx)
		if err != nil {
			return fmt.Errorf("pop: %w", err)
		}
		if len(msgs) == 0 {
			continue // the long poll expired with an empty queue
		}

		for _, msg := range msgs {
			seen, known := pending[msg.TransactionID]
			if !known {
				return fmt.Errorf("consumed a message nobody pushed: %s", msg.TransactionID)
			}
			if seen {
				return fmt.Errorf("message %s was delivered twice", msg.TransactionID)
			}
			pending[msg.TransactionID] = true
			consumed++
			fmt.Printf("consumed %v (%s)\n", msg.Data["orderId"], msg.TransactionID)
		}

		// The ack is the commit. It advances this consumer group's cursor past
		// the messages; without it the lease expires and they are redelivered.
		acks, err := client.Ack(ctx, msgs, true, queen.AckOptions{ConsumerGroup: consumerGroup})
		if err != nil {
			return fmt.Errorf("ack: %w", err)
		}
		// A rejected ack still arrives as HTTP 200 with success=false on its
		// item, so a nil error is not enough. Check every entry.
		for i, ack := range acks {
			if !ack.Success {
				return fmt.Errorf("ack rejected for %s: %s", msgs[i].TransactionID, ack.Error)
			}
		}
		fmt.Printf("acked %d message(s)\n", len(msgs))
	}

	if consumed != messageCount {
		return fmt.Errorf("expected %d messages, consumed %d", messageCount, consumed)
	}
	for txID, seen := range pending {
		if !seen {
			return fmt.Errorf("message %s was never delivered", txID)
		}
	}

	// The cursor is committed to the end of the queue, so a further pop must
	// come back empty. Wait(false) returns immediately with whatever is there.
	leftovers, err := client.Queue(queueName).
		Group(consumerGroup).
		Batch(messageCount).
		Wait(false).
		Pop(ctx)
	if err != nil {
		return fmt.Errorf("drain check: %w", err)
	}
	if len(leftovers) != 0 {
		return fmt.Errorf("expected nothing left after acking, got %d message(s)", len(leftovers))
	}

	// Housekeeping, not part of the lesson: drop the queue this run created.
	if _, err := client.Queue(queueName).Delete().Execute(ctx); err != nil {
		return fmt.Errorf("delete queue: %w", err)
	}

	return nil
}