PubSubClient

import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
import io.ktor.client.HttpClient
import io.ktor.client.request.post
import io.ktor.client.request.put
import io.ktor.content.TextContent
import io.ktor.http.ContentType
import java.util.*

class PubSubClient(label: String) {

    private val projectId: String = System.getenv("PUBSUB_PROJECT_ID")
    private val pubSubHostPort: String = System.getenv("PUBSUB_EMULATOR_HOST")
    private val topic = "$label-topic"
    private val subscription = "$label-subscription"
    private val baseUrl = "http://$pubSubHostPort/v1"
    private val topicPath = "projects/$projectId/topics/$topic"
    private val subscriptionPath = "projects/$projectId/subscriptions/$subscription"

    private val client = HttpClient()

    private suspend fun createTopic() {
        client.put<Unit>("$baseUrl/$topicPath")
    }

    private suspend fun createSubscription() {
        client.put<Unit>("$baseUrl/$subscriptionPath") {
            body = TextContent(
                """{"topic":"$topicPath","ackDeadlineSeconds":10}""",
                ContentType.Application.Json
            )
        }
    }

    suspend fun createTopicAndSubscription() {
        createTopic()
        createSubscription()
    }

    private val objectMapper = jacksonObjectMapper()

    suspend fun publishMessage(byteArray: ByteArray) {

        data class Message(
            val data: String
        )

        data class Messages(
            val messages: Collection<Message>
        )

        val messages = Messages(listOf(Message(data = Base64.getEncoder().encodeToString(byteArray))))
        client.post<Unit>("$baseUrl/$topicPath:publish") {
            body = TextContent(objectMapper.writeValueAsString(messages),ContentType.Application.Json)
        }
    }
}