Google Pub/Sub #

dart pub add googleapis googleapis_auth http

# pubspec.yaml
dependencies:
  googleapis: ^12.0.0
  googleapis_auth: ^1.5.0
  http: ^1.2.0

import 'package:googleapis_auth/auth_io.dart';
import 'package:googleapis/pubsub/v1.dart';
import 'dart:io';
import 'dart:convert';

// Method 1: Service Account JSON Key File
Future<PubsubApi> buatKlienDariFile(String pathKeyFile) async {
  final keyJson = jsonDecode(await File(pathKeyFile).readAsString())
      as Map<String, dynamic>;

  final credentials = ServiceAccountCredentials.fromJson(keyJson);
  final scopes = [PubsubApi.cloudPlatformScope];

  final httpClient = await clientViaServiceAccount(credentials, scopes);
  return PubsubApi(httpClient);
}

// Method 2: Application Default Credentials (ADC)
// Works automatically on GCE, Cloud Run, GKE, Cloud Functions
// Local: run `gcloud auth application-default login`
Future<PubsubApi> buatKlienADC() async {
  final scopes = [PubsubApi.cloudPlatformScope];
  final httpClient = await clientViaApplicationDefaultCredentials(
    scopes: scopes,
  );
  return PubsubApi(httpClient);
}

// Method 3: Emulator for local development
// Run: gcloud beta emulators pubsub start
Future<PubsubApi> buatKlienEmulator() async {
  // Set the environment variable: PUBSUB_EMULATOR_HOST=localhost:8085
  final emulatorHost = Platform.environment['PUBSUB_EMULATOR_HOST'];
  if (emulatorHost != null) {
    // The emulator doesn't need authentication
    print('Using the Pub/Sub emulator: $emulatorHost');
  }
  return buatKlienADC(); // ADC will automatically use the emulator if the env var is set
}

import 'package:googleapis/pubsub/v1.dart';

const projectId = 'my-gcp-project';

String topicPath(String nama) => 'projects/$projectId/topics/$nama';
String subPath(String nama) => 'projects/$projectId/subscriptions/$nama';

Future<void> setupInfrastruktur(PubsubApi pubsub) async {
  // Create a topic
  try {
    await pubsub.projects.topics.create(
      Topic(
        name: topicPath('order-events'),
        messageRetentionDuration: '604800s',  // keep for 7 days (for replay)
        labels: {'environment': 'production', 'team': 'platform'},
      ),
      topicPath('order-events'),
    );
    print('Topic created successfully');
  } on DetailedApiRequestError catch (e) {
    if (e.status == 409) {
      print('Topic already exists, skipping');
    } else {
      rethrow;
    }
  }

  // Create the Dead Letter Topic first
  try {
    await pubsub.projects.topics.create(
      Topic(name: topicPath('order-events-dlq')),
      topicPath('order-events-dlq'),
    );
  } on DetailedApiRequestError catch (e) {
    if (e.status != 409) rethrow;
  }

  // Create a Pull Subscription
  await pubsub.projects.subscriptions.create(
    Subscription(
      name: subPath('order-processor'),
      topic: topicPath('order-events'),
      ackDeadlineSeconds: 60,          // 60 seconds to acknowledge
      enableMessageOrdering: false,    // true if ordering per key is needed
      deadLetterPolicy: DeadLetterPolicy(
        deadLetterTopic: topicPath('order-events-dlq'),
        maxDeliveryAttempts: 5,        // try 5 times before going to the DLQ
      ),
      retryPolicy: RetryPolicy(
        minimumBackoff: '10s',
        maximumBackoff: '300s',  // exponential backoff up to 5 minutes
      ),
      messageRetentionDuration: '604800s',
      expirationPolicy: ExpirationPolicy(ttl: '2592000s'),  // 30 days
    ),
    subPath('order-processor'),
  );

  // Create a Push Subscription — Pub/Sub sends to an HTTP endpoint
  await pubsub.projects.subscriptions.create(
    Subscription(
      name: subPath('webhook-notifier'),
      topic: topicPath('order-events'),
      ackDeadlineSeconds: 30,
      pushConfig: PushConfig(
        pushEndpoint: 'https://myapp.example.com/pubsub/webhook',
        oidcToken: OidcToken(
          serviceAccountEmail: '[email protected]',
        ),
      ),
    ),
    subPath('webhook-notifier'),
  );

  // List all topics
  final topics = await pubsub.projects.topics.list('projects/$projectId');
  print('Topics: ${topics.topics?.map((t) => t.name).toList()}');
}

import 'package:googleapis/pubsub/v1.dart';
import 'dart:convert';

class GCPPublisher {
  final PubsubApi _pubsub;
  final String _topicPath;

  GCPPublisher({required PubsubApi pubsub, required String topicName})
      : _pubsub = pubsub,
        _topicPath = 'projects/$projectId/topics/$topicName';

  // Send a single message
  Future<String> kirim({
    required Map<String, dynamic> data,
    Map<String, String>? attributes,
    String? orderingKey,  // for ordered messages per key
  }) async {
    final pesanBytes = base64Encode(utf8.encode(jsonEncode(data)));

    final response = await _pubsub.projects.topics.publish(
      PublishRequest(
        messages: [
          PubsubMessage(
            data: pesanBytes,           // data must be in base64 format
            attributes: {
              'tipeEvent': data['tipe'] as String? ?? 'unknown',
              'versi': '1.0',
              'timestamp': DateTime.now().toUtc().toIso8601String(),
              ...?attributes,
            },
            orderingKey: orderingKey,   // null = no ordering guarantee
          ),
        ],
      ),
      _topicPath,
    );

    final messageId = response.messageIds?.first ?? '';
    print('Message sent: $messageId');
    return messageId;
  }

  // Batch publish — up to 1.000 messages or 10MB per request
  Future<List<String>> kirimBatch(List<Map<String, dynamic>> pesanList) async {
    final messages = pesanList.map((data) => PubsubMessage(
      data: base64Encode(utf8.encode(jsonEncode(data))),
      attributes: {
        'tipeEvent': data['tipe'] as String? ?? 'unknown',
      },
    )).toList();

    final response = await _pubsub.projects.topics.publish(
      PublishRequest(messages: messages),
      _topicPath,
    );

    return response.messageIds ?? [];
  }
}

// Usage
Future<void> main() async {
  final pubsub = await buatKlienADC();
  final publisher = GCPPublisher(pubsub: pubsub, topicName: 'order-events');

  await publisher.kirim(
    data: {
      'tipe': 'order_dibuat',
      'idOrder': 'ORD-001',
      'idPengguna': 'USR-123',
      'total': 150_000,
    },
    attributes: {'sumber': 'checkout-service'},
    orderingKey: 'USR-123',  // all events for this user are sent in order
  );
}

import 'package:googleapis/pubsub/v1.dart';
import 'dart:convert';

class GCPPullSubscriber {
  final PubsubApi _pubsub;
  final String _subscriptionPath;
  bool _jalan = true;

  GCPPullSubscriber({
    required PubsubApi pubsub,
    required String subscriptionName,
  })  : _pubsub = pubsub,
        _subscriptionPath = 'projects/$projectId/subscriptions/$subscriptionName';

  Future<void> mulai(
    Future<void> Function(Map<String, dynamic> data, Map<String, String> attributes) handler,
  ) async {
    print('Subscriber started at: $_subscriptionPath');

    while (_jalan) {
      try {
        // Pull messages — max 100 per request
        final response = await _pubsub.projects.subscriptions.pull(
          PullRequest(
            maxMessages: 100,
            returnImmediately: false,  // wait until there are messages (blocking)
          ),
          _subscriptionPath,
        );

        final messages = response.receivedMessages ?? [];
        if (messages.isEmpty) continue;

        print('Received ${messages.length} messages');

        final ackIds = <String>[];
        final nackIds = <String>[];

        await Future.wait(
          messages.map((received) async {
            final msg = received.message!;
            try {
              // Decode data from base64
              final rawData = utf8.decode(base64Decode(msg.data!));
              final data = jsonDecode(rawData) as Map<String, dynamic>;
              final attributes = msg.attributes ?? {};

              print('Processing: ${msg.messageId} (type: ${attributes['tipeEvent']})');
              await handler(data, Map<String, String>.from(attributes));

              ackIds.add(received.ackId!);  // mark as successful
            } catch (e) {
              print('Failed to process ${msg.messageId}: $e');
              nackIds.add(received.ackId!);  // mark as failed for retry
            }
          }),
        );

        // Acknowledge all successful ones
        if (ackIds.isNotEmpty) {
          await _pubsub.projects.subscriptions.acknowledge(
            AcknowledgeRequest(ackIds: ackIds),
            _subscriptionPath,
          );
        }

        // Modify the ack deadline for failures — shorten for faster retries
        if (nackIds.isNotEmpty) {
          await _pubsub.projects.subscriptions.modifyAckDeadline(
            ModifyAckDeadlineRequest(
              ackIds: nackIds,
              ackDeadlineSeconds: 0,  // 0 = immediately available again for retry
            ),
            _subscriptionPath,
          );
        }

      } catch (e) {
        print('Error while pulling: $e');
        await Future.delayed(Duration(seconds: 5));
      }
    }
  }

  void hentikan() => _jalan = false;
}

// Usage
Future<void> main() async {
  final pubsub = await buatKlienADC();
  final subscriber = GCPPullSubscriber(
    pubsub: pubsub,
    subscriptionName: 'order-processor',
  );

  await subscriber.mulai((data, attributes) async {
    print('Event: ${attributes['tipeEvent']}');
    print('Data: $data');
    await prosesEventOrder(data);
  });
}

import 'package:shelf/shelf.dart';
import 'package:shelf/shelf_io.dart' as shelf_io;
import 'dart:convert';

// Webhook handler that receives pushes from Pub/Sub
Handler buatWebhookHandler() {
  return (Request request) async {
    if (request.method != 'POST') {
      return Response.methodNotAllowed();
    }

    try {
      final body = await request.readAsString();
      final envelope = jsonDecode(body) as Map<String, dynamic>;

      // The Pub/Sub push envelope has this structure:
      // {
      //   "message": {
      //     "data": "<base64-encoded-data>",
      //     "attributes": {...},
      //     "messageId": "...",
      //     "publishTime": "..."
      //   },
      //   "subscription": "projects/.../subscriptions/..."
      // }
      final message = envelope['message'] as Map<String, dynamic>;
      final rawData = utf8.decode(base64Decode(message['data'] as String));
      final data = jsonDecode(rawData) as Map<String, dynamic>;
      final attributes = (message['attributes'] as Map<String, dynamic>?)
          ?.map((k, v) => MapEntry(k, v as String)) ??
          {};

      print('Push received: ${message['messageId']}');
      print('Subscription: ${envelope['subscription']}');

      await prosesEventOrder(data);

      // Return 2xx so Pub/Sub knows the message succeeded (auto-acknowledge)
      return Response.ok('OK');

    } catch (e) {
      print('Error processing push: $e');
      // Return non-2xx so Pub/Sub resends it (retry)
      return Response.internalServerError(body: e.toString());
    }
  };
}

// Run the webhook server
Future<void> main() async {
  final handler = Pipeline()
      .addMiddleware(logRequests())
      .addHandler(buatWebhookHandler());

  final server = await shelf_io.serve(handler, '0.0.0.0', 8080);
  print('Webhook server: http://${server.address.host}:${server.port}');
}

// Subscription with a filter — only receive specific messages
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('payment-processor'),
    topic: topicPath('order-events'),
    ackDeadlineSeconds: 60,
    // Filter using the Pub/Sub filter language
    filter: 'attributes.tipeEvent = "pembayaran_dikonfirmasi"',
  ),
  subPath('payment-processor'),
);

// More complex filters
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('premium-order-processor'),
    topic: topicPath('order-events'),
    filter: 'attributes.tipeEvent = "order_dibuat" AND attributes.levelPengguna = "premium"',
  ),
  subPath('premium-order-processor'),
);

// Filters to exclude specific types
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('non-payment-processor'),
    topic: topicPath('order-events'),
    filter: 'NOT attributes.tipeEvent = "pembayaran_dikonfirmasi"',
  ),
  subPath('non-payment-processor'),
);

// For subscriptions with ordering, enable it on the subscription
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('ordered-processor'),
    topic: topicPath('order-events'),
    enableMessageOrdering: true,  // required to receive ordering keys
  ),
  subPath('ordered-processor'),
);

// Publishers must also send with ordering keys
// All messages with the same ordering key are guaranteed to be ordered
await publisher.kirim(
  data: {'tipe': 'order_dibuat', 'idOrder': 'ORD-100'},
  orderingKey: 'customer-C001',  // all events for this customer are ordered
);

await publisher.kirim(
  data: {'tipe': 'pembayaran_dikonfirmasi', 'idOrder': 'ORD-100'},
  orderingKey: 'customer-C001',  // received after order_dibuat for this customer
);