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';

// Cara 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);
}

// Cara 2: Application Default Credentials (ADC)
// Bekerja secara otomatis di GCE, Cloud Run, GKE, Cloud Functions
// Lokal: jalankan `gcloud auth application-default login`
Future<PubsubApi> buatKlienADC() async {
  final scopes = [PubsubApi.cloudPlatformScope];
  final httpClient = await clientViaApplicationDefaultCredentials(
    scopes: scopes,
  );
  return PubsubApi(httpClient);
}

// Cara 3: Emulator untuk development lokal
// Jalankan: gcloud beta emulators pubsub start
Future<PubsubApi> buatKlienEmulator() async {
  // Set environment variable: PUBSUB_EMULATOR_HOST=localhost:8085
  final emulatorHost = Platform.environment['PUBSUB_EMULATOR_HOST'];
  if (emulatorHost != null) {
    // Emulator tidak butuh autentikasi
    print('Menggunakan Pub/Sub emulator: $emulatorHost');
  }
  return buatKlienADC(); // ADC akan otomatis gunakan emulator jika env var di-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 {
  // Buat topic
  try {
    await pubsub.projects.topics.create(
      Topic(
        name: topicPath('order-events'),
        messageRetentionDuration: '604800s',  // simpan 7 hari (untuk replay)
        labels: {'environment': 'production', 'team': 'platform'},
      ),
      topicPath('order-events'),
    );
    print('Topic berhasil dibuat');
  } on DetailedApiRequestError catch (e) {
    if (e.status == 409) {
      print('Topic sudah ada, lewati');
    } else {
      rethrow;
    }
  }

  // Buat Dead Letter Topic terlebih dahulu
  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;
  }

  // Buat Pull Subscription
  await pubsub.projects.subscriptions.create(
    Subscription(
      name: subPath('order-processor'),
      topic: topicPath('order-events'),
      ackDeadlineSeconds: 60,          // 60 detik untuk acknowledge
      enableMessageOrdering: false,    // true jika butuh ordering per key
      deadLetterPolicy: DeadLetterPolicy(
        deadLetterTopic: topicPath('order-events-dlq'),
        maxDeliveryAttempts: 5,        // coba 5 kali sebelum ke DLQ
      ),
      retryPolicy: RetryPolicy(
        minimumBackoff: '10s',
        maximumBackoff: '300s',  // exponential backoff hingga 5 menit
      ),
      messageRetentionDuration: '604800s',
      expirationPolicy: ExpirationPolicy(ttl: '2592000s'),  // 30 hari
    ),
    subPath('order-processor'),
  );

  // Buat Push Subscription — Pub/Sub kirim ke 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 semua topic
  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';

  // Kirim satu pesan
  Future<String> kirim({
    required Map<String, dynamic> data,
    Map<String, String>? attributes,
    String? orderingKey,  // untuk pesan terurut per key
  }) async {
    final pesanBytes = base64Encode(utf8.encode(jsonEncode(data)));

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

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

  // Batch publish — hingga 1.000 pesan atau 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 ?? [];
  }
}

// Penggunaan
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',  // semua event user ini dikirim terurut
  );
}

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 mulai pada: $_subscriptionPath');

    while (_jalan) {
      try {
        // Pull pesan — max 100 per request
        final response = await _pubsub.projects.subscriptions.pull(
          PullRequest(
            maxMessages: 100,
            returnImmediately: false,  // tunggu hingga ada pesan (blocking)
          ),
          _subscriptionPath,
        );

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

        print('Menerima ${messages.length} pesan');

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

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

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

              ackIds.add(received.ackId!);  // tandai berhasil
            } catch (e) {
              print('Gagal memproses ${msg.messageId}: $e');
              nackIds.add(received.ackId!);  // tandai gagal untuk retry
            }
          }),
        );

        // Acknowledge semua yang berhasil
        if (ackIds.isNotEmpty) {
          await _pubsub.projects.subscriptions.acknowledge(
            AcknowledgeRequest(ackIds: ackIds),
            _subscriptionPath,
          );
        }

        // Modify ack deadline untuk yang gagal — perpendek agar lebih cepat retry
        if (nackIds.isNotEmpty) {
          await _pubsub.projects.subscriptions.modifyAckDeadline(
            ModifyAckDeadlineRequest(
              ackIds: nackIds,
              ackDeadlineSeconds: 0,  // 0 = segera tersedia kembali untuk retry
            ),
            _subscriptionPath,
          );
        }

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

  void hentikan() => _jalan = false;
}

// Penggunaan
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';

// Handler webhook yang menerima push dari 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>;

      // Pub/Sub push envelope memiliki struktur:
      // {
      //   "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 diterima: ${message['messageId']}');
      print('Subscription: ${envelope['subscription']}');

      await prosesEventOrder(data);

      // Kembalikan 2xx agar Pub/Sub tahu pesan berhasil (auto-acknowledge)
      return Response.ok('OK');

    } catch (e) {
      print('Error memproses push: $e');
      // Kembalikan non-2xx agar Pub/Sub kirim ulang (retry)
      return Response.internalServerError(body: e.toString());
    }
  };
}

// Jalankan server webhook
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 dengan filter — hanya terima pesan tertentu
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('payment-processor'),
    topic: topicPath('order-events'),
    ackDeadlineSeconds: 60,
    // Filter menggunakan Pub/Sub filter language
    filter: 'attributes.tipeEvent = "pembayaran_dikonfirmasi"',
  ),
  subPath('payment-processor'),
);

// Filter lebih kompleks
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'),
);

// Filter untuk exclude tipe tertentu
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'),
);

// Untuk subscription dengan ordering, enable di subscription
await pubsub.projects.subscriptions.create(
  Subscription(
    name: subPath('ordered-processor'),
    topic: topicPath('order-events'),
    enableMessageOrdering: true,  // wajib untuk menerima ordering key
  ),
  subPath('ordered-processor'),
);

// Publisher juga harus mengirim dengan ordering key
// Semua pesan dengan ordering key yang sama dijamin terurut
await publisher.kirim(
  data: {'tipe': 'order_dibuat', 'idOrder': 'ORD-100'},
  orderingKey: 'customer-C001',  // semua event customer ini terurut
);

await publisher.kirim(
  data: {'tipe': 'pembayaran_dikonfirmasi', 'idOrder': 'ORD-100'},
  orderingKey: 'customer-C001',  // diterima setelah order_dibuat untuk customer ini
);