amqp091

tinox.core.amqp091, split out of the bundled stdlib as a standalone tinox-central package.

Publish consume

import tinox.core.amqp091
import tinox.core.encoding

class Main
{
    fnc main() -> Int32
    {
        const connection: AmqpConnection091 = AmqpConnection091::connect("localhost", 5672, "/", "guest", "guest")
        const amqpChannel: AmqpChannel091 = AmqpChannel091::open(connection)

        const queue: String = amqpChannel.declareQueue("jobs", true, false, false)
        const body: List<Int64> = Encoding::toUtf8("hello")
        amqpChannel.publish("", queue, body, "text/plain")

        amqpChannel.consume(queue)
        const msg: AmqpMessage091 = amqpChannel.nextMessage()
        println(msg.body.len().toString() + " bytes received")
        amqpChannel.ack(msg.deliveryTag)

        connection.close()
        return 0
    }
}

class Amqp091

Methods

class AmqpChannel091

Fields

Methods

class AmqpConfirmResult091

Fields

class AmqpConnection091

Fields

Methods

class AmqpFrame091

Fields

class AmqpMessage091

Fields

class AmqpMethodFrame091

Fields

fn tinoxBytesToString(bytes: List<Int64>) → String

class AmqpReader091

Fields

Methods

class AmqpWriter091

Fields

Methods

class IdempotencyWindow091

* A bounded, in-memory window of recently handled idempotency keys (typically * AmqpMessage091.messageId). It suppresses redeliveries within one process -- not across * restarts or instances; durable de-duplication needs a database unique key. The window keeps * at most `capacity` keys (the oldest are forgotten first), so it cannot grow without bound. * * if (!this.window.firstSeen(msg.messageId)) { return } // duplicate: acknowledged, not handled * ... handle ... * (on failure the handler throws and the key is forgotten by the caller via forget())

Fields

Methods

class MessageFailure091

* Failure policy of an @Amqp091Consumer handler, used by the generated receive loop * (@MessageFailure(maxAttempts, backoffMs[, "dead-letter-queue"]) on the @OnMessage method). * * A delivery is acknowledged only when the handler returned normally. After a failure the * handler runs again in the same process, waiting backoffMs * 2^(attempt-1) (capped at 60 s) * in between, until maxAttempts have failed. Then the message is settled, never acknowledged * as if handled: * - with a dead-letter queue: it is published there (default exchange, routing key = the * queue name; the body and content type are kept, headers are not) and only then * acknowledged; if that publish cannot be done the delivery is rejected with requeue, so * the message stays in the broker instead of being lost; * - without one: basic.reject(requeue=false) -- the broker discards it or dead-letters it * through the queue's own x-dead-letter-exchange. Broker semantics are not replaced. * Retries are in-process: a crash or restart leaves the delivery unacknowledged and the broker * redelivers it (redelivered == true) -- at-least-once, never exactly-once. Handlers must be * idempotent; IdempotencyWindow091 helps for the redelivery window.

Methods