Skip to main content

AMQPQueue

@webda/amqp

Class: AMQPQueue<T, K>

Defined in: amqp/src/queue.service.ts:52

Implements a Queue stored in AMQP

Webda Modda​

Extends​

  • Queue<T, K>

Type Parameters​

T​

T = any

K​

K extends AMQPQueueParameters = AMQPQueueParameters

Constructors​

Constructor​

new AMQPQueue<T, K>(name, params): AMQPQueue<T, K>

Defined in: core/lib/services/service.d.ts:72

Service

Parameters​

name​

string

The name of the service

params​

K

The parameters block define in the configuration file

Returns​

AMQPQueue<T, K>

Inherited from​

Queue<T, K>.constructor

Properties​

_compiledCapabilities​

protected _compiledCapabilities: Record<string, any>

Defined in: core/lib/services/iservice.d.ts:38

Capabilities detected at compile-time from @WebdaCapability-tagged interfaces.

Populated during Service.resolve by reading the service's entry in webda.module.json. Each key is a capability name (e.g., "request-filter"), and the value is an empty object {} by default. Override getCapabilities to provide capability-specific configuration or to conditionally disable capabilities.

See​

getCapabilities

Inherited from​

Queue._compiledCapabilities

_timeout​

protected _timeout: Timeout

Defined in: core/lib/queues/queueservice.d.ts:70

Current timeout handler

Inherited from​

Queue._timeout

[WEBDA_EVENTS]​

[WEBDA_EVENTS]: object

Defined in: core/lib/services/service.d.ts:50

Set the Webda events here

Inherited from​

Queue.[WEBDA_EVENTS]

channel?​

optional channel?: Channel

Defined in: amqp/src/queue.service.ts:53

conn?​

optional conn?: ChannelModel

Defined in: amqp/src/queue.service.ts:54

delayer​

protected delayer: WaitDelayer

Defined in: core/lib/queues/queueservice.d.ts:82

Delayer

Inherited from​

Queue.delayer

eventPrototype​

eventPrototype: new () => T

Defined in: core/lib/queues/queueservice.d.ts:87

Returns​

T

Inherited from​

Queue.eventPrototype

failedIterations​

protected failedIterations: number

Defined in: core/lib/queues/queueservice.d.ts:78

Current pause instance

Inherited from​

Queue.failedIterations

idleDelayer​

protected idleDelayer: WaitDelayer

Defined in: core/lib/queues/queueservice.d.ts:86

Delayer between two receives returning no message

Inherited from​

Queue.idleDelayer

logger​

protected logger: Logger

Defined in: core/lib/services/service.d.ts:59

Logger with class context

Inherited from​

Queue.logger

metrics​

protected metrics: object

Defined in: core/lib/queues/pubsubservice.d.ts:10

errors​

errors: Counter

messages_pending​

messages_pending: Gauge

messages_received​

messages_received: Counter

messages_sent​

messages_sent: Counter

processing_duration​

processing_duration: Histogram

Inherited from​

Queue.metrics

name​

readonly name: string

Defined in: core/lib/services/iservice.d.ts:18

Inherited from​

Queue.name

parameters​

readonly parameters: K

Defined in: core/lib/services/iservice.d.ts:19

Inherited from​

Queue.parameters

createConfiguration?​

static optional createConfiguration?: (params) => any

Defined in: core/lib/services/iservice.d.ts:23

Create configuration set by the application on load

Parameters​

params​

any

Returns​

any

Inherited from​

Queue.createConfiguration

filterConfiguration?​

static optional filterConfiguration?: (params) => any

Defined in: core/lib/services/iservice.d.ts:27

Create configuration set by the application on load

Parameters​

params​

any

Returns​

any

Inherited from​

Queue.filterConfiguration

Parameters​

static Parameters: typeof ServiceParameters

Defined in: core/lib/services/service.d.ts:46

Service parameters

Inherited from​

Queue.Parameters

Methods​

__clean()​

__clean(): Promise<void>

Defined in: amqp/src/queue.service.ts:270

Purge the queue, used by tests

Returns​

Promise<void>

Overrides​

Queue.__clean

addListener()​

addListener<Key>(eventName, listener): this

Defined in: core/lib/events/asynceventemitter.d.ts:80

Type Parameters​

Key​

Key extends never

Parameters​

eventName​

Key

the event name

listener​

(event) => void | Promise<void>

the event listener

Returns​

this

this for chaining

See​

EventEmitter.addListener

Inherited from​

Queue.addListener

addRoute()​

protected addRoute(url, methods, executer, openapi?, override?): void

Defined in: core/lib/services/service.d.ts:177

Add a route dynamicaly

Parameters​

url​

string

of the route can contains dynamic part like {uuid}

methods​

HttpMethodType[]

the HTTP methods

executer​

Function

Method to execute for this route

openapi?​

OpenAPIWebdaDefinition

the OpenAPI specification

override?​

boolean

whether to override existing

Returns​

void

Inherited from​

Queue.addRoute

authorizeClientEvent()​

authorizeClientEvent(_event, _context): boolean

Defined in: core/lib/services/service.d.ts:153

Authorize a public event subscription

Parameters​

_event​

string

the event name

_context​

OperationContext

the execution context

Returns​

boolean

true if the condition is met

Inherited from​

Queue.authorizeClientEvent

computeParameters()​

computeParameters(): void

Defined in: core/lib/services/service.d.ts:77

Used to compute or derivate input parameter to attribute

Returns​

void

Deprecated​

Inherited from​

Queue.computeParameters

consume()​

consume(callback, eventPrototype?): CancelablePromise

Defined in: amqp/src/queue.service.ts:150

Work the queue, with the messages pushed by the broker when push is enabled

The callback runs as each message is delivered; the message is acknowledged once it succeeds. When it fails, the message is rejected and requeued after the workerDelayer delay. Cancelling stops the delivery, waits for the callbacks in progress and closes the channel: the messages not acknowledged go back to the queue.

Parameters​

callback​

(event) => Promise<void>

the callback function

eventPrototype?​

new () => T

the event prototype

Returns​

CancelablePromise

the consumer, resolved once cancelled

Overrides​

Queue.consume

consumerReceiveMessage()​

protected consumerReceiveMessage(): Promise<{ items: number; speed: number; }>

Defined in: core/lib/queues/queueservice.d.ts:109

Receive and process message from the queue

Returns​

Promise<{ items: number; speed: number; }>

the result

Inherited from​

Queue.consumerReceiveMessage

deleteMessage()​

deleteMessage(id): Promise<void>

Defined in: amqp/src/queue.service.ts:126

Parameters​

id​

string

the receipt handle returned by receiveMessage

Returns​

Promise<void>

Overrides​

Queue.deleteMessage

emit()​

emit<Key>(event, data): Promise<void>

Defined in: core/lib/services/service.d.ts:204

Emit the event with data and wait for Promise to finish if listener returned a Promise

Type Parameters​

Key​

Key extends never

Parameters​

event​

Key

the event name

data​

object[Key]

the data to process

Returns​

Promise<void>

Inherited from​

Queue.emit

getCapabilities()​

getCapabilities(): Record<string, any>

Defined in: core/lib/services/iservice.d.ts:60

Return the capabilities of this service.

By default returns capabilities detected at compile-time from

Returns​

Record<string, any>

the result

Webda Capability-tagged​

interfaces in webda.module.json.

Override to disable capabilities based on configuration:

getCapabilities() {
  const caps = super.getCapabilities();
  if (!this.parameters.enabled) delete caps["request-filter"];
  return caps;
}

Inherited from​

Queue.getCapabilities

getChannel()​

protected getChannel(): Channel

Defined in: amqp/src/queue.service.ts:60

Returns​

Channel

the active AMQP channel

Throws​

CodeError when the service is not connected

getClientEvents()​

getClientEvents(): string[]

Defined in: core/lib/services/service.d.ts:144

Return the events that an external system can subscribe to

Returns​

string[]

the list of results

Inherited from​

Queue.getClientEvents

getIdleDelay()​

getIdleDelay(emptyReceives): number

Defined in: core/lib/queues/queueservice.d.ts:119

Delay before the next receive after consecutive receives returned no message, before the maxIdleDelay cap

Parameters​

emptyReceives​

number

number of consecutive empty receives, starting at 1

Returns​

number

the delay in milliseconds

Inherited from​

Queue.getIdleDelay

getMaxConsumers()​

getMaxConsumers(): number

Defined in: core/lib/queues/queueservice.d.ts:130

Return the max consumers for the queue

It is overridable so if a queue can retrieve several message at once it can just use the worker // and several messages at once

SQS for example will return this.parameters.maxConsumers / 10

Returns​

number

the result number

Inherited from​

Queue.getMaxConsumers

getMaxListeners()​

getMaxListeners(): number

Defined in: core/lib/events/asynceventemitter.d.ts:84

Returns​

number

Inherited from​

Queue.getMaxListeners

getMetric()​

getMetric<T>(type, configuration): T

Defined in: core/lib/services/service.d.ts:138

Add service name label

Type Parameters​

T​

T = Counter<string> | Histogram<string> | Gauge<string>

Parameters​

type​

CustomConstructor<T, [MetricConfiguration<T>]>

the type to look up

configuration​

MetricConfiguration<T>

the configuration

Returns​

T

the result

Inherited from​

Queue.getMetric

getName()​

getName(): string

Defined in: core/lib/services/service.d.ts:209

Get service name

Returns​

string

the result string

Inherited from​

Queue.getName

getOpenApiReplacements()​

getOpenApiReplacements(): any

Defined in: core/lib/services/service.d.ts:182

Return variables for replacement in openapi

Returns​

any

the result

Inherited from​

Queue.getOpenApiReplacements

getOperationId()​

getOperationId(id): string | undefined

Defined in: core/lib/services/service.d.ts:167

If undefined is returned it cancel the operation registration

Parameters​

id​

string

the identifier

Returns​

string | undefined

the result

Inherited from​

Queue.getOperationId

getParameters()​

getParameters(): K

Defined in: core/lib/services/service.d.ts:82

Get the service parameters

Returns​

K

the result

Inherited from​

Queue.getParameters

getService()​

getService<T>(name): ServicesMap[T]

Defined in: core/lib/services/service.d.ts:131

Get a service by name

Type Parameters​

T​

T extends keyof ServicesMap

Parameters​

name​

T

the name to use

Returns​

ServicesMap[T]

the result map

Deprecated​

Use useService, might reconsider

Inherited from​

Queue.getService

getState()​

getState(): ServiceStates

Defined in: core/lib/services/service.d.ts:55

Get the current state

Returns​

ServiceStates

the result

Inherited from​

Queue.getState

getUrl()​

getUrl(url, _methods): string

Defined in: core/lib/services/service.d.ts:161

Return the full path url based on parameters

Parameters​

url​

string

relative url to service

_methods​

HttpMethodType[]

in case we need filtering (like Store)

Returns​

string

absolute url or undefined if need to skip the Route

Inherited from​

Queue.getUrl

init()​

init(): Promise<AMQPQueue<T, K>>

Defined in: amqp/src/queue.service.ts:71

Returns​

Promise<AMQPQueue<T, K>>

this service

Overrides​

Queue.init

initMetrics()​

initMetrics(): void

Defined in: core/lib/queues/pubsubservice.d.ts:20

Returns​

void

Inherited from​

Queue.initMetrics

initOperations()​

initOperations(): void

Defined in: core/lib/services/service.d.ts:186

Init the operations from

Returns​

void

Operation​

decorators on this service

Inherited from​

Queue.initOperations

listeners()​

listeners(eventName): Function[]

Defined in: core/lib/events/asynceventemitter.d.ts:139

Get all listeners for an event

Parameters​

eventName​

never

the event name

Returns​

Function[]

the list of results

Inherited from​

Queue.listeners

loadCapabilities()​

protected loadCapabilities(): void

Defined in: core/lib/services/service.d.ts:120

Load capabilities from webda.module.json metadata into _compiledCapabilities.

Called during resolve after dependency injection. Reads the service's type name from parameters, looks it up in the application's module metadata (moddas or beans section), and populates _compiledCapabilities with an empty object for each declared capability name.

Fails silently if the application is not available (e.g., in unit tests where services are instantiated without a full application context).

Returns​

void

Example​

// If webda.module.json contains:
// { "moddas": { "MyApp/HawkService": { "capabilities": ["request-filter", "cors-filter"] } } }
// Then after resolve(), this.getCapabilities() returns:
// { "request-filter": {}, "cors-filter": {} }

See​

getCapabilities

Inherited from​

Queue.loadCapabilities

log()​

log(level, ...args): void

Defined in: core/lib/services/service.d.ts:226

Parameters​

level​

WorkerLogLevel

to log

args​

...any[]

additional arguments

Returns​

void

Inherited from​

Queue.log

off()​

off<Key>(eventName, listener): this

Defined in: core/lib/events/asynceventemitter.d.ts:116

Type Parameters​

Key​

Key extends never

Parameters​

eventName​

Key

the event name

listener​

(event) => void

the event listener

Returns​

this

this for chaining

See​

EventEmitter.off

Inherited from​

Queue.off

on()​

on<Key>(eventName, listener): this

Defined in: core/lib/events/asynceventemitter.d.ts:102

Type Parameters​

Key​

Key extends never

Parameters​

eventName​

Key

the event name

listener​

(event) => void

the event listener

Returns​

this

this for chaining

See​

EventEmitter.once

Inherited from​

Queue.on

once()​

once<Key>(eventName, listener): this

Defined in: core/lib/events/asynceventemitter.d.ts:95

Type Parameters​

Key​

Key extends never

Parameters​

eventName​

Key

the event name

listener​

(event) => void

the event listener

Returns​

this

this for chaining

See​

EventEmitter.once

Inherited from​

Queue.once

receiveMessage()​

receiveMessage<L>(proto?): Promise<MessageReceipt<L>[]>

Defined in: amqp/src/queue.service.ts:109

Type Parameters​

L​

L

Parameters​

proto?​

new () => L

optional prototype to rehydrate the payload into

Returns​

Promise<MessageReceipt<L>[]>

the received message if any

Overrides​

Queue.receiveMessage

removeAllListeners()​

removeAllListeners<Key>(eventName?): this

Defined in: core/lib/events/asynceventemitter.d.ts:122

Type Parameters​

Key​

Key extends never

Parameters​

eventName?​

Key

the event name

Returns​

this

this for chaining

See​

EventEmitter.removeAllListeners

Inherited from​

Queue.removeAllListeners

removeListener()​

removeListener<Key>(eventName, listener): this

Defined in: core/lib/events/asynceventemitter.d.ts:109

Type Parameters​

Key​

Key extends never

Parameters​

eventName​

Key

the event name

listener​

(event) => void

the event listener

Returns​

this

this for chaining

See​

EventEmitter.removeListener

Inherited from​

Queue.removeListener

resolve()​

resolve(): this

Defined in: core/lib/queues/queueservice.d.ts:103

Create the delayer

Returns​

this

this for chaining

Inherited from​

Queue.resolve

sendMessage()​

sendMessage(event): Promise<void>

Defined in: amqp/src/queue.service.ts:99

Parameters​

event​

T

the event to enqueue

Returns​

Promise<void>

Overrides​

Queue.sendMessage

setMaxListeners()​

setMaxListeners(n): this

Defined in: core/lib/events/asynceventemitter.d.ts:88

Parameters​

n​

number

Returns​

this

Inherited from​

Queue.setMaxListeners

size()​

size(): Promise<number>

Defined in: amqp/src/queue.service.ts:134

Returns​

Promise<number>

the number of messages ready in the queue

Overrides​

Queue.size

stop()​

stop(): Promise<void>

Defined in: amqp/src/queue.service.ts:83

Close the AMQP connection

Returns​

Promise<void>

Overrides​

Queue.stop

toJSON()​

toJSON(): string

Defined in: core/lib/services/service.d.ts:191

Prevent service to be serialized

Returns​

string

the result

Inherited from​

Queue.toJSON

toString()​

toString(): string

Defined in: core/lib/services/service.d.ts:91

Return service representation

Returns​

string

the result

Inherited from​

Queue.toString

unserialize()​

unserialize<L>(data, proto?): L

Defined in: core/lib/queues/pubsubservice.d.ts:32

Unserialize into class

Type Parameters​

L​

L

Parameters​

data​

string

the data to process

proto?​

new () => L

the proto

Returns​

L

the result

Inherited from​

Queue.unserialize