kiss_gcp_pubsub_queue 0.2.0

A Google Cloud Pub/Sub backend for kiss_queue providing SQS-style queue semantics

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.