Order Processing - Choreography - Shipment Service

shipment_service.kt

class ShipmentService(private val customerAddresses: MutableMap<Int, String> = mutableMapOf()) {
    @KafkaListener(topics = [PAYMENTS])
    fun onOrder(order: Order) {
        if (order.status !== PAID) return

        sendShipment(order, customerAddresses[order.customerId] ?: throw RuntimeException("No address for customer ${order.customerId}"))
        sendMessage(SHIPMENTS, order.withStatus(SHIPPED))
    }

    @KafkaListener(topics = [CUSTOMERS])
    fun onCustomer(customer: Customer) {
        customerAddresses[customer.id] = customer.address
    }
}