Vela - NestJS for the Edge

The QueueModule provides a managed QueueService for producing messages, and the @QueueConsumer() decorator for consuming them.

Setup

Import the module and configure the binding name:

import { Module } from '@velajs/vela';
import { QueueModule } from '@velajs/cloudflare';

@Module({
  imports: [QueueModule.forRoot({ binding: 'MY_QUEUE' })],
})
class AppModule {}

Add the queue producer to wrangler.toml:

[[queues.producers]]
binding = "MY_QUEUE"
queue = "my-queue"

[[queues.consumers]]
queue = "my-queue"

Producing Messages

Inject QueueService to send messages:

import { Controller, Post, Body } from '@velajs/vela';
import { QueueService } from '@velajs/cloudflare';

@Controller('tasks')
class TasksController {
  constructor(private readonly queue: QueueService) {}

  @Post()
  async enqueue(@Body() body: { task: string }) {
    await this.queue.send(body);
    return { queued: true };
  }

  @Post('batch')
  async enqueueBatch(@Body() body: { tasks: string[] }) {
    await this.queue.sendBatch(body.tasks.map((t) => ({ body: t })));
    return { queued: body.tasks.length };
  }
}

Consuming Messages

Use the @QueueConsumer() decorator on a method to handle incoming batches:

import { Injectable } from '@velajs/vela';
import { QueueConsumer } from '@velajs/cloudflare';

@Injectable()
class TaskProcessor {
  @QueueConsumer('my-queue')
  async process(batch: MessageBatch) {
    for (const msg of batch.messages) {
      console.log('Processing:', msg.body);
      msg.ack();
    }
  }
}

Wire the queue handler on the application:

const app = CloudflareFactory.create(AppModule);
export default app;

The app.queue(batch, env, ctx) handler is automatically registered and delegates to the decorated consumer method.

Methods

Method	Description
`send(message)`	Send a single message to the queue.
`sendBatch(messages)`	Send multiple messages in a single call. Each entry has a `body` and optional `contentType`.

Queues

Setup

Producing Messages

Consuming Messages

Methods

On this page