kiss_gcp_pubsub_queue

A Google Cloud Pub/Sub backend for kiss_queue — part of the KISS (Keep It Simple, Stupid) family of libraries.

🎯 Purpose

kiss_gcp_pubsub_queue provides a production-ready Google Cloud Pub/Sub implementation of the kiss_queue interface. It brings SQS-style queue semantics to GCP, including visibility timeouts, dead letter queues, and FIFO ordering with deduplication.

Just queues on GCP. No ceremony. No complexity.

✨ Features

  • 🔄 Standard Queue: High throughput, unordered, at-least-once delivery
  • 📋 FIFO Queue: Ordered message delivery per group with client-side deduplication
  • ⏱️ Visibility Timeout: Configurable invisibility window via ack deadline management
  • 💀 Dead Letter Queues: Automatic routing after max delivery attempts
  • 🔌 Serialization Support: Pluggable serialization for any data format
  • 🧪 Comprehensive Testing: 113 tests covering all functionality
  • 🏗️ Dual Client Architecture: Uses gcloud for publishing, googleapis for pull and control plane

🚀 Quick Start

Basic Usage

import 'package:gcloud/pubsub.dart';
import 'package:googleapis/pubsub/v1.dart' as pubsub_api;
import 'package:kiss_gcp_pubsub_queue/kiss_gcp_pubsub_queue.dart';

void main() async {
  // Create authenticated clients (see Authentication section)
  final pubsub = PubSub(client, projectId);
  final pubsubApi = pubsub_api.PubsubApi(client);

  // Create admin and factory
  final admin = PubSubAdmin(pubsubApi, projectId: projectId);
  final factory = PubSubQueueFactory<String, String>(
    pubsub: pubsub,
    admin: admin,
  );

  // Create a queue
  final queue = await factory.createQueue('my-queue');

  // Enqueue a message
  await queue.enqueuePayload('Hello, World!');

  // Dequeue and process
  final message = await queue.dequeue();
  if (message != null) {
    print('Received: ${message.payload}');
    await queue.acknowledge(message.id!);
  }

  // Cleanup
  await factory.dispose();
}

Serialization Example

import 'dart:convert';
import 'package:kiss_queue/kiss_queue.dart';
import 'package:kiss_gcp_pubsub_queue/kiss_gcp_pubsub_queue.dart';

class Order {
  final String orderId;
  final double amount;
  Order(this.orderId, this.amount);

  Map<String, dynamic> toJson() => {'orderId': orderId, 'amount': amount};
  static Order fromJson(Map<String, dynamic> json) =>
      Order(json['orderId'], json['amount']);
}

class OrderSerializer implements MessageSerializer<Order, String> {
  @override
  String serialize(Order payload) => jsonEncode(payload.toJson());

  @override
  Order deserialize(String data) => Order.fromJson(jsonDecode(data));
}

void main() async {
  final factory = PubSubQueueFactory<Order, String>(
    pubsub: pubsub,
    admin: admin,
    serializer: OrderSerializer(),
  );

  final queue = await factory.createQueue('order-queue');

  // Enqueue with automatic serialization
  await queue.enqueuePayload(Order('ORD-123', 99.99));

  // Dequeue with automatic deserialization
  final message = await queue.dequeue();
  if (message != null) {
    print('Order: ${message.payload.orderId}'); // Fully typed Order object
    await queue.acknowledge(message.id!);
  }
}

FIFO Queue with Ordering

final fifoQueue = await factory.createQueue(
  'ordered-queue',
  configuration: const PubSubQueueConfiguration(
    isFifo: true,
    groupId: 'my-group',
  ),
);

// Messages with the same groupId are delivered in order
await fifoQueue.enqueuePayload('First');
await fifoQueue.enqueuePayload('Second');
await fifoQueue.enqueuePayload('Third');

Dead Letter Queue

// Create DLQ first
final dlq = await factory.createQueue('my-queue-dlq');

// Create main queue with DLQ
final queue = await factory.createQueue(
  'my-queue',
  configuration: const PubSubQueueConfiguration(
    maxReceiveCount: 5, // Move to DLQ after 5 failed attempts
  ),
  deadLetterQueue: dlq,
);

🏗️ Architecture

Dual Client Design

This package uses two Google Cloud client libraries:

Library Purpose Operations
gcloud Publishing publish messages to topics
googleapis Pull & control plane pull, acknowledge, modifyAckDeadline, create/delete topics and subscriptions, configure DLQ

The gcloud package provides a high-level API for publishing, while googleapis handles all pull and administrative operations, ensuring access to real Pub/Sub ack IDs for reliable message lifecycle management.

Resource Naming

Queues are created with the following Pub/Sub resources:

Resource Naming Pattern
Topic kiss-{queueName}
FIFO Topic kiss-{queueName}-fifo
Subscription kiss-{queueName}-sub
DLQ Topic kiss-{queueName}-dlq
DLQ Subscription kiss-{queueName}-dlq-sub

Message Lifecycle

  1. Enqueue: Publish message to Pub/Sub topic (with optional serialization)
  2. Dequeue: Pull message from subscription (becomes invisible via ack deadline)
  3. Process: Handle the message in your application
  4. Acknowledge: Confirm successful processing (removes from subscription)
  5. Reject: Mark as failed (requeue for retry or let DLQ policy handle it)

FIFO Deduplication

Pub/Sub doesn't natively deduplicate messages, so FIFO queues use client-side deduplication via a DedupeStore. The default implementation uses kiss_repository's InMemoryRepository.

// Custom dedupe store (e.g., Redis-backed for distributed systems)
final queue = PubSubQueue<String, String>(
  topic: topic,
  admin: admin,
  configuration: const PubSubQueueConfiguration(isFifo: true),
  subscriptionName: 'my-sub',
  dedupeStore: MyRedisDedupeStore(),
);

⚙️ Configuration

PubSubQueueConfiguration

const config = PubSubQueueConfiguration(
  visibilityTimeout: Duration(seconds: 30),  // Ack deadline
  maxReceiveCount: 5,                        // Max attempts before DLQ
  isFifo: false,                             // Enable ordering + deduplication
  groupId: null,                             // Message group for FIFO
  dedupeTtl: Duration(minutes: 5),           // Deduplication window
);
Option Default Description
visibilityTimeout 30 seconds How long a message is invisible after being received
maxReceiveCount 5 Max delivery attempts before sending to DLQ
isFifo false Enable ordered delivery and deduplication
groupId null Message group for FIFO ordering
contentBasedDeduplication false Use SHA-256 hash of payload for deduplication instead of message ID
dedupeTtl 5 minutes Deduplication window for FIFO queues

🔐 Authentication

Using Application Default Credentials

import 'package:googleapis_auth/auth_io.dart';

final client = await clientViaApplicationDefaultCredentials(
  scopes: [pubsub_api.PubsubApi.pubsubScope],
);

Using Service Account

import 'package:googleapis_auth/auth_io.dart';

final credentials = ServiceAccountCredentials.fromJson(jsonCredentials);
final client = await clientViaServiceAccount(
  credentials,
  [pubsub_api.PubsubApi.pubsubScope],
);

🧪 Testing

The package includes 113 comprehensive tests:

# Run all tests
dart test

# Run specific test file
dart test test/queue_test.dart

Test Coverage

  • ✅ Queue operations (enqueue, dequeue, acknowledge, reject)
  • ✅ FIFO ordering and deduplication
  • ✅ Dead letter queue routing
  • ✅ Visibility timeout (ack deadline) management
  • ✅ Serialization/deserialization
  • ✅ Factory queue creation and caching
  • ✅ Admin operations (create/delete topics and subscriptions)

🛠️ Installation

Add to your pubspec.yaml:

dependencies:
  kiss_gcp_pubsub_queue: ^0.2.0

Then run:

dart pub get

📦 Dependencies

Package Purpose
kiss_queue Core queue interface
gcloud Publishing messages to topics
googleapis Pull, acknowledge, and control plane operations
kiss_repository Dedupe store implementation

🤝 Contributing

We welcome contributions! Please see our contributing guidelines for details.

Running Tests

# Run all tests
dart test

# Run with coverage
dart test --coverage=coverage

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🌟 Why kiss_gcp_pubsub_queue?

  • Simple: Unified kiss_queue interface, no Pub/Sub complexity exposed
  • Reliable: SQS-style semantics with visibility timeouts and DLQ
  • Flexible: Works with any serialization format
  • Performant: Optimized for high throughput
  • Testable: 113 comprehensive tests included
  • Production Ready: Built for real-world GCP deployments

Perfect for microservices, event-driven architectures, and any GCP application that needs reliable async message processing.

Built with ❤️ by the WAMF team. Part of the KISS family of simple, focused Dart packages.

Libraries

kiss_gcp_pubsub_queue