Google Pub/Sub #

// build.gradle.kts
dependencies {
    // Google Cloud Pub/Sub
    implementation("com.google.cloud:google-cloud-pubsub:1.127.2")

    // kotlinx.serialization
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")

    // Coroutines
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.0")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-guava:1.8.0")  // Guava Future → coroutine
}

// How ADC works (automatically, without code):
// 1. The GOOGLE_APPLICATION_CREDENTIALS environment variable (path to a service account JSON)
// 2. gcloud auth application-default login (for local development)
// 3. A Service Account attached to GCE/GKE/Cloud Run

// For local development:
// $ gcloud auth application-default login
// $ export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json

// Or create credentials explicitly in code:
import com.google.auth.oauth2.GoogleCredentials
import com.google.auth.oauth2.ServiceAccountCredentials
import java.io.FileInputStream

fun createCredentials(): GoogleCredentials {
    val serviceAccountPath = System.getenv("GOOGLE_APPLICATION_CREDENTIALS")
        ?: throw IllegalStateException("GOOGLE_APPLICATION_CREDENTIALS is not set")

    return ServiceAccountCredentials.fromStream(FileInputStream(serviceAccountPath))
}

import com.google.cloud.pubsub.v1.TopicAdminClient
import com.google.cloud.pubsub.v1.SubscriptionAdminClient
import com.google.pubsub.v1.*

object PubSubAdmin {
    private val PROJECT_ID = System.getenv("GOOGLE_CLOUD_PROJECT") ?: "my-project"

    fun topicName(name: String) = ProjectTopicName.of(PROJECT_ID, name).toString()
    fun subscriptionName(name: String) = ProjectSubscriptionName.of(PROJECT_ID, name).toString()

    fun createTopic(name: String): String {
        val topicNameStr = topicName(name)

        TopicAdminClient.create().use { admin ->
            return try {
                val topic = admin.createTopic(topicNameStr)
                println("Topic created: ${topic.name}")
                topic.name
            } catch (e: com.google.api.gax.rpc.AlreadyExistsException) {
                println("Topic already exists: $topicNameStr")
                topicNameStr
            }
        }
    }

    fun createPullSubscription(subName: String, topicId: String): String {
        val subNameStr = subscriptionName(subName)
        val topicNameStr = topicName(topicId)

        val subscriptionRequest = Subscription.newBuilder()
            .setName(subNameStr)
            .setTopic(topicNameStr)
            .setAckDeadlineSeconds(60)          // 60 seconds to process and ack
            .setMessageRetentionDuration(         // keep messages for up to 7 days
                com.google.protobuf.Duration.newBuilder().setSeconds(7 * 24 * 3600).build()
            )
            .setRetainAckedMessages(false)        // delete acked messages
            .build()

        SubscriptionAdminClient.create().use { admin ->
            return try {
                val sub = admin.createSubscription(subscriptionRequest)
                println("Pull Subscription created: ${sub.name}")
                sub.name
            } catch (e: com.google.api.gax.rpc.AlreadyExistsException) {
                println("Subscription already exists: $subNameStr")
                subNameStr
            }
        }
    }

    fun createPushSubscription(subName: String, topicId: String, endpointUrl: String): String {
        val subNameStr = subscriptionName(subName)
        val topicNameStr = topicName(topicId)

        val pushConfig = PushConfig.newBuilder()
            .setPushEndpoint(endpointUrl)  // the URL that will receive POSTs from Pub/Sub
            .build()

        val subscriptionRequest = Subscription.newBuilder()
            .setName(subNameStr)
            .setTopic(topicNameStr)
            .setPushConfig(pushConfig)
            .setAckDeadlineSeconds(60)
            .build()

        SubscriptionAdminClient.create().use { admin ->
            val sub = admin.createSubscription(subscriptionRequest)
            println("Push Subscription created: ${sub.name} → $endpointUrl")
            return sub.name
        }
    }

    // Subscription with a filter — only receive messages with certain attributes
    fun createSubscriptionWithFilter(subName: String, topicId: String, filter: String): String {
        // Example filter: attributes.status = "NEW"
        // or: hasPrefix(attributes.tipe, "pesanan.")
        val subNameStr = subscriptionName(subName)

        val subscriptionRequest = Subscription.newBuilder()
            .setName(subNameStr)
            .setTopic(topicName(topicId))
            .setFilter(filter)
            .setAckDeadlineSeconds(30)
            .build()

        SubscriptionAdminClient.create().use { admin ->
            val sub = admin.createSubscription(subscriptionRequest)
            println("Subscription with filter '$filter' created: ${sub.name}")
            return sub.name
        }
    }
}

import com.google.cloud.pubsub.v1.Publisher
import com.google.protobuf.ByteString
import com.google.pubsub.v1.PubsubMessage
import com.google.pubsub.v1.TopicName
import kotlinx.serialization.Serializable
import kotlinx.serialization.json.Json

@Serializable
data class OrderEvent(
    val orderId: String,
    val userId: Long,
    val total: Double,
    val status: String,
    val timestamp: Long = System.currentTimeMillis()
)

class PubSubPublisher(
    private val projectId: String,
    private val topicId: String
) {
    private val json = Json { ignoreUnknownKeys = true }
    private val topicName = TopicName.of(projectId, topicId)

    // Use use{} to make sure the publisher is shut down and all messages are flushed
    fun <T> withPublisher(block: Publisher.() -> T): T {
        val publisher = Publisher.newBuilder(topicName)
            .build()
        return try {
            publisher.block()
        } finally {
            publisher.shutdown()
            publisher.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)
        }
    }

    fun send(event: OrderEvent): String {
        return withPublisher {
            val payload = json.encodeToString(OrderEvent.serializer(), event)
            val data = ByteString.copyFromUtf8(payload)

            val message = PubsubMessage.newBuilder()
                .setData(data)
                .putAttributes("tipe", "OrderEvent")      // attributes for filtering
                .putAttributes("versi", "1.0")
                .putAttributes("status", event.status)
                .build()

            val future = publish(message)
            val messageId = future.get()  // block until confirmation
            println("Message sent — ID: $messageId")
            messageId
        }
    }

    // Send with an ordering key — guarantees order per key
    fun sendWithOrdering(event: OrderEvent, orderingKey: String): String {
        val publisher = Publisher.newBuilder(topicName)
            .setEnableMessageOrdering(true)  // must be enabled on the publisher
            .build()

        return try {
            val payload = json.encodeToString(OrderEvent.serializer(), event)
            val message = PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8(payload))
                .setOrderingKey(orderingKey)  // messages with the same key are guaranteed ordered
                .putAttributes("userId", event.userId.toString())
                .build()

            publisher.publish(message).get()
        } finally {
            publisher.shutdown()
            publisher.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)
        }
    }

    // Send a batch — the publisher does internal batching automatically,
    // but we can send many at once and wait for all to finish
    fun sendMany(events: List<OrderEvent>): List<String> {
        return withPublisher {
            val futures = events.map { event ->
                val message = PubsubMessage.newBuilder()
                    .setData(ByteString.copyFromUtf8(
                        json.encodeToString(OrderEvent.serializer(), event)
                    ))
                    .putAttributes("orderId", event.orderId)
                    .build()
                publish(message)
            }
            futures.map { it.get() }  // wait for all to finish
        }
    }
}

import com.google.cloud.pubsub.v1.AckReplyConsumer
import com.google.cloud.pubsub.v1.MessageReceiver
import com.google.cloud.pubsub.v1.Subscriber
import com.google.pubsub.v1.ProjectSubscriptionName
import com.google.pubsub.v1.PubsubMessage

class PullSubscriber(
    private val projectId: String,
    private val subscriptionId: String
) {
    private val json = Json { ignoreUnknownKeys = true }
    private val subscriptionName = ProjectSubscriptionName.of(projectId, subscriptionId)

    // Async subscriber — the callback is invoked for each new message
    fun startAsync(): Subscriber {
        val receiver = MessageReceiver { message: PubsubMessage, consumer: AckReplyConsumer ->
            try {
                processMessage(message)
                consumer.ack()   // tell Pub/Sub the message was processed successfully
            } catch (e: Exception) {
                println("Failed to process: ${e.message}")
                consumer.nack()  // Pub/Sub will redeliver after the ack deadline
            }
        }

        val subscriber = Subscriber.newBuilder(subscriptionName, receiver)
            .setFlowControlSettings(
                com.google.api.gax.batching.FlowControlSettings.newBuilder()
                    .setMaxOutstandingElementCount(100)  // max 100 unacked messages
                    .setMaxOutstandingRequestBytes(10 * 1024 * 1024)  // max 10MB
                    .build()
            )
            .build()

        subscriber.startAsync().awaitRunning()
        println("Subscriber '$subscriptionId' started listening...")
        return subscriber
    }

    private fun processMessage(message: PubsubMessage) {
        val payload = message.data.toStringUtf8()
        val type = message.attributesMap["tipe"]
        val messageId = message.messageId

        println("Message received — ID: $messageId, type: $type")

        when (type) {
            "OrderEvent" -> {
                val event = json.decodeFromString(OrderEvent.serializer(), payload)
                println("Processing order: ${event.orderId} — ${event.status}")
                // Business logic...
            }
            else -> println("Unknown type: $type")
        }
    }
}

fun main() {
    val subscriber = PullSubscriber("my-project", "pesanan-email").startAsync()

    // Run until interrupted
    try {
        Thread.currentThread().join()
    } catch (e: InterruptedException) {
        subscriber.stopAsync()
        println("Subscriber stopped")
    }
}

import io.ktor.server.application.*
import io.ktor.server.request.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import kotlinx.serialization.Serializable
import java.util.Base64

@Serializable
data class PubSubPushPayload(
    val message: PubSubPushMessage,
    val subscription: String
)

@Serializable
data class PubSubPushMessage(
    val data: String,                           // base64-encoded payload
    val messageId: String,
    val publishTime: String,
    val attributes: Map<String, String> = emptyMap()
)

// Ktor handler for receiving pushes from Pub/Sub
fun Application.pubSubPushHandler() {
    val json = Json { ignoreUnknownKeys = true }

    routing {
        post("/pubsub/push") {
            val body = call.receiveText()

            val payload = runCatching {
                json.decodeFromString(PubSubPushPayload.serializer(), body)
            }.getOrElse {
                call.respond(io.ktor.http.HttpStatusCode.BadRequest, "Invalid body")
                return@post
            }

            // Decode the data from base64
            val dataBytes = Base64.getDecoder().decode(payload.message.data)
            val messageText = String(dataBytes)

            println("Push received — ID: ${payload.message.messageId}")
            println("Attributes: ${payload.message.attributes}")
            println("Data: $messageText")

            // Process the message
            try {
                val event = json.decodeFromString(OrderEvent.serializer(), messageText)
                processPushEvent(event)

                // HTTP 200/204 = ack (Pub/Sub won't redeliver)
                call.respond(io.ktor.http.HttpStatusCode.NoContent)
            } catch (e: Exception) {
                println("Failed to process: ${e.message}")
                // HTTP 4xx/5xx = nack (Pub/Sub will redeliver after backoff)
                call.respond(io.ktor.http.HttpStatusCode.InternalServerError, "Failed to process")
            }
        }
    }
}

fun processPushEvent(event: OrderEvent) {
    println("Push: processing order ${event.orderId}")
}

fun createSubscriptionWithDlt(
    subName: String,
    topicId: String,
    dltName: String,
    maxAttempts: Int = 5
): String {
    // Make sure the DLT exists
    PubSubAdmin.createTopic(dltName)

    val dltNameStr = PubSubAdmin.topicName(dltName)
    val subNameStr = PubSubAdmin.subscriptionName(subName)

    val deadLetterPolicy = DeadLetterPolicy.newBuilder()
        .setDeadLetterTopic(dltNameStr)
        .setMaxDeliveryAttempts(maxAttempts)  // try max N times before the DLT
        .build()

    val subscriptionRequest = Subscription.newBuilder()
        .setName(subNameStr)
        .setTopic(PubSubAdmin.topicName(topicId))
        .setDeadLetterPolicy(deadLetterPolicy)
        .setAckDeadlineSeconds(30)
        .build()

    SubscriptionAdminClient.create().use { admin ->
        val sub = admin.createSubscription(subscriptionRequest)
        println("Subscription with DLT created: ${sub.name}")
        println("Dead Letter Topic: $dltNameStr (max $maxAttempts attempts)")
        return sub.name
    }
}

# docker-compose.yml
services:
  pubsub-emulator:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: gcloud beta emulators pubsub start --host-port=0.0.0.0:8085
    ports:
      - "8085:8085"

# Run the emulator
docker compose up -d

# Set the environment variable to point the SDK at the emulator
export PUBSUB_EMULATOR_HOST=localhost:8085

// In Kotlin code, the SDK automatically detects PUBSUB_EMULATOR_HOST
// If that env var is set, all Pub/Sub calls are directed to the emulator

// You can also override the endpoint explicitly
import com.google.api.gax.core.NoCredentialsProvider
import com.google.api.gax.grpc.GrpcTransportChannel
import com.google.api.gax.rpc.FixedTransportChannelProvider
import io.grpc.ManagedChannelBuilder

fun createEmulatorPublisher(projectId: String, topicId: String): Publisher {
    val channel = ManagedChannelBuilder
        .forTarget("localhost:8085")
        .usePlaintext()
        .build()

    val channelProvider = FixedTransportChannelProvider.create(
        GrpcTransportChannel.create(channel)
    )

    return Publisher.newBuilder(TopicName.of(projectId, topicId))
        .setChannelProvider(channelProvider)
        .setCredentialsProvider(NoCredentialsProvider.create())
        .build()
}

// Publisher — enable ordering and set an ordering key
fun sendWithOrdering() {
    val publisher = Publisher.newBuilder(TopicName.of("my-project", "pesanan-events"))
        .setEnableMessageOrdering(true)
        .build()

    val events = listOf(
        OrderEvent("ORD-001", 1L, 100_000.0, "CREATED"),
        OrderEvent("ORD-001", 1L, 100_000.0, "PAID"),
        OrderEvent("ORD-001", 1L, 100_000.0, "SHIPPED"),
        OrderEvent("ORD-001", 1L, 100_000.0, "DONE")
    )

    val json = Json { ignoreUnknownKeys = true }

    events.forEach { event ->
        val message = PubsubMessage.newBuilder()
            .setData(ByteString.copyFromUtf8(json.encodeToString(OrderEvent.serializer(), event)))
            .setOrderingKey(event.orderId)  // all ORD-001 events are ordered
            .build()
        publisher.publish(message).get()
        println("Sent: ${event.orderId} — ${event.status}")
    }

    publisher.shutdown()
    publisher.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)
}

// The subscription must also have ordering enabled
// enableMessageOrdering = true in the subscription settings