メインコンテンツまでスキップ

Queue message handler

Queue topic の message を backend service method で処理します。​

Queue Message Handler を使用する理由​

application が、注文の作成、file の upload、payment の完了などの event を queue に publish し、backend でそれぞれを確実に処理する必要がある場合に使用します。

@onQueueMessage では、method を decorate して deploy するだけです。

Backend code
@onQueueMessage<OrderEvent>('order-events')
async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
  const { orderId, customerEmail, status } = request.message;
  // Email the customer an order confirmation using your own email helper.
  await this.sendOrderConfirmationEmail(customerEmail, orderId, status);
}

polling も routing logic も不要です。topic に message が到着した瞬間に Squid が handler を呼び出します。

概要​

Queue message handler は、指定した topic に message が publish されるたびに自動的に実行される、@onQueueMessage（Python では @on_queue_message）で decorate された backend method です。built-in Squid queue と、Kafka などの external queue integration の両方をサポートします。

Queue message handler を使用する場合​

ユースケース推奨
Queue topic に publish された message を処理する✅ Queue message handler
Database change に反応するTriggers を使用
Client から function を呼び出すExecutables を使用
Schedule に基づいて code を実行するSchedulers を使用
External service に HTTP endpoint を公開するWebhooks を使用

仕組み​

  1. SquidService を拡張する class で、method を @onQueueMessage() で decorate します
  2. topic name と、必要に応じて integration ID を指定します
  3. Squid が deploy 時に handler を検出して登録します
  4. topic に message が到着すると、Squid は QueueMessageRequest object を指定して method を呼び出します
  5. handler は synchronous にすることも、Promise を返すこともできます

クイックスタート​

前提条件​

  • squid init で初期化された Squid backend project
  • NPM からインストールされた @squidcloud/backend package

ステップ 1: Handler を作成する​

SquidService を拡張する service class を作成し、handler method を追加します。

Backend code
import { SquidService, onQueueMessage, QueueMessageRequest } from '@squidcloud/backend';

interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

export class OrderService extends SquidService {
  @onQueueMessage<OrderEvent>('order-events')
  async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    const { orderId, customerEmail, status } = request.message;
    // Email the customer an order confirmation using your own email helper.
    await this.sendOrderConfirmationEmail(customerEmail, orderId, status);
  }
}

ステップ 2: Service を export する​

service が service index file から export されていることを確認します。

Backend code
export * from './example-service';

ステップ 3: Backend を開始または deploy する​

ローカル開発では、Squid CLI を使用して backend をローカルで実行します。

squid start

cloud に deploy するには、backend の deployを参照してください。

ステップ 4: 検証する​

client または別の service から topic に message を publish し、Squid Console log を確認して handler が呼び出されたことを確認します。

コアコンセプト​

Decorator​

decorator は 2 つの parameter を受け取ります。

ParameterType必須説明
topicNamestringはいsubscribe する queue topic の name
integrationIdstringいいえqueue の integration ID。default は built-in Squid queue integration です
Backend code
// Built-in queue: integrationId defaults to 'built_in_queue'
@onQueueMessage('order-events')

// External integration (e.g. Kafka)
@onQueueMessage('order-events', 'kafka')

QueueMessageRequest​

handler に渡される QueueMessageRequest object には以下が含まれます。

PropertyType説明
messageTtype 付けされた message payload
topicNamestringmessage の送信先 topic の name
integrationIdstringqueue の integration ID

TypeScript では、generic type parameter T により publisher から handler まで end-to-end の type safety を得られます。Python では QueueMessageRequest は TypedDict のため、dictionary-style lookup で field にアクセスします。

Backend code
interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

@onQueueMessage<OrderEvent>('order-events')
async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
  // request.message is typed as OrderEvent
  const { orderId, status } = request.message;
}

External queue integration の使用​

integrationId を省略すると、Squid の built-in queue が使用されます。明示的な integration ID を指定すると、同じ handler pattern を Kafka などの external queue system で使用できます。

Backend code
import { SquidService, onQueueMessage, QueueMessageRequest } from '@squidcloud/backend';

interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

export class OrderService extends SquidService {
  // Built-in queue
  @onQueueMessage<OrderEvent>('order-events')
  async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    console.log('Built-in queue event:', request.message);
  }

  // External Kafka integration
  @onQueueMessage<OrderEvent>('order-events', 'kafka')
  async handleKafkaOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    console.log(`Kafka event on topic "${request.topicName}":`, request.message);
  }
}

Error Handling​

handler が error を throw すると、その error は Squid Console に log されます。failure を適切に処理するため、logic を try/catch（または try/except）block でラップしてください。

Backend code
@onQueueMessage<OrderEvent>('order-events')
async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
  try {
    await this.processOrder(request.message);
  } catch (error) {
    console.error(`Failed to process order ${request.message.orderId}:`, error);
  }
}

ベストプラクティス​

  1. Message payload に type を付ける。 TypeScript では QueueMessageRequest<MyType> generic parameter を、Python では TypedDict subclass を使用して、message body に type-safe に access します。

  2. Handler 内で error を処理する。 logic を try/catch（または try/except）block でラップし、error を log に記録して、1 つの不正な message が暗黙的に失敗しないようにします。

  3. Handler の焦点を絞る。 handler は 1 つのことを行うべきです。message が複数の workflow を trigger する必要がある場合は、すべての logic を handler に記述するのではなく、他の method または service に delegate してください。

  4. idempotency を考慮して設計する。 message は、まれに複数回配信されることがあります。duplicate message を処理しても同じ result になるよう handler を設計してください。

コード例​

Queue message を external service に転送する​

Backend code
import { SquidService, onQueueMessage, QueueMessageRequest } from '@squidcloud/backend';

interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

export class OrderService extends SquidService {
  @onQueueMessage<OrderEvent>('order-events')
  async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    // Forward each order event to your fulfillment service.
    await this.forwardToFulfillment(request.message);
  }
}

複数の queue integration からの message を処理する​

Backend code
import { SquidService, onQueueMessage, QueueMessageRequest } from '@squidcloud/backend';

interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

export class OrderService extends SquidService {
  @onQueueMessage<OrderEvent>('order-events')
  async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    await this.processOrder(request.message);
  }

  @onQueueMessage<OrderEvent>('order-events', 'kafka')
  async handleKafkaOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    await this.processOrder(request.message);
  }

  private async processOrder(event: OrderEvent): Promise<void> {
    // Email the customer an order confirmation using your own email helper.
    await this.sendOrderConfirmationEmail(event.customerEmail, event.orderId, event.status);
  }
}

Full-Stack Example​

この例では、完全な flow を示します。client が order event を publish し、backend handler がそれを処理します。

Backend: handler を登録する​

Backend code
import { SquidService, onQueueMessage, QueueMessageRequest, secureTopic } from '@squidcloud/backend';

interface OrderEvent {
  orderId: string;
  customerEmail: string;
  status: 'placed' | 'shipped' | 'delivered';
}

export class OrderService extends SquidService {
  @secureTopic('order-events', 'write')
  allowOrderEventPublish(): boolean {
    return !!this.getUserAuth();
  }

  @onQueueMessage<OrderEvent>('order-events')
  async handleOrderEvent(request: QueueMessageRequest<OrderEvent>): Promise<void> {
    const { orderId, customerEmail, status } = request.message;
    console.log(`Received order event: ${orderId} → ${status}`);
    // Email the customer an order confirmation using your own email helper.
    await this.sendOrderConfirmationEmail(customerEmail, orderId, status);
  }
}

Client: message を publish する​

Client code
const orderEvent = { orderId: 'order-123', customerEmail: 'jane@example.com', status: 'placed' };
await squid.queue('order-events').produce([orderEvent]);

produce が呼び出されると、Squid は message を topic に配信し、backend handler が自動的に実行されます。

関連項目​