Consumer Event Contract

import {
  Consumer,
  MessageHandler,
  Messages,
  BucketEventType,
} from "@celerity-sdk/core";
import type { ConsumerMessage, EventResult, BucketEvent } from "@celerity-sdk/core";

@Consumer("uploads-bucket")
class BucketEventHandler {
  @MessageHandler()
  async handle(@Messages() messages: ConsumerMessage[]): Promise<EventResult> {
    for (const msg of messages) {
      const event: BucketEvent = JSON.parse(msg.body);

      if (msg.eventType === BucketEventType.Created) {
        console.log(
          `New object in ${msg.sourceName}: ${event.key} (${event.size} bytes)`
        );
      } else if (msg.eventType === BucketEventType.Deleted) {
        console.log(`Object deleted from ${msg.sourceName}: ${event.key}`);
      }
    }
    return { success: true };
  }
}

import json
from celerity_sdk import consumer, message_handler, BucketEventType, EventResult

@consumer("uploads-bucket")
class BucketEventHandler:
    @message_handler()
    async def handle(self, messages):
        for msg in messages:
            event = json.loads(msg.body)

            if msg.event_type == BucketEventType.CREATED:
                print(
                    f"New object in {msg.source_name}: {event['key']} ({event['size']} bytes)"
                )
            elif msg.event_type == BucketEventType.DELETED:
                print(f"Object deleted from {msg.source_name}: {event['key']}")

        return EventResult(success=True)

import {
  Consumer,
  MessageHandler,
  Messages,
  DatastoreEventType,
} from "@celerity-sdk/core";
import type {
  ConsumerMessage,
  EventResult,
  DatastoreEvent,
} from "@celerity-sdk/core";

@Consumer("orders-table")
class DatastoreEventHandler {
  @MessageHandler()
  async handle(@Messages() messages: ConsumerMessage[]): Promise<EventResult> {
    for (const msg of messages) {
      const event: DatastoreEvent = JSON.parse(msg.body);

      switch (msg.eventType) {
        case DatastoreEventType.Inserted:
          console.log(`New item in ${msg.sourceName}:`, event.keys);
          console.log("Item:", event.newItem);
          break;
        case DatastoreEventType.Modified:
          console.log(`Item modified in ${msg.sourceName}:`, event.keys);
          console.log("Before:", event.oldItem);
          console.log("After:", event.newItem);
          break;
        case DatastoreEventType.Removed:
          console.log(`Item removed from ${msg.sourceName}:`, event.keys);
          break;
      }
    }
    return { success: true };
  }
}

import json
from celerity_sdk import consumer, message_handler, DatastoreEventType, EventResult

@consumer("orders-table")
class DatastoreEventHandler:
    @message_handler()
    async def handle(self, messages):
        for msg in messages:
            event = json.loads(msg.body)

            if msg.event_type == DatastoreEventType.INSERTED:
                print(f"New item in {msg.source_name}:", event["keys"])
                print("Item:", event["newItem"])
            elif msg.event_type == DatastoreEventType.MODIFIED:
                print(f"Item modified in {msg.source_name}:", event["keys"])
                print("Before:", event["oldItem"])
                print("After:", event["newItem"])
            elif msg.event_type == DatastoreEventType.REMOVED:
                print(f"Item removed from {msg.source_name}:", event["keys"])

        return EventResult(success=True)