---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: support@cybus.io
# ----------------------------------------------------------------------------
# Confluent Cloud Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the bidirectional integration between
  Connectware and Confluent Cloud (Example)

metadata:
  name: Confluent Cloud Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  bootstrapServer:
    description: Bootstrap server of your Confluent Cloud cluster
    type: string
    default: pkc-00000.eu-central-1.aws.confluent.cloud:9092

  confluentApiKey:
    description: Confluent Cloud cluster API key
    type: string

  confluentApiSecret:
    description: Confluent Cloud cluster API secret
    type: string

  produceTopic:
    description: Confluent Cloud topic that receives the shop floor data
    type: string
    default: factory.machine-data

  consumeTopic:
    description: Confluent Cloud topic that Connectware consumes into MQTT
    type: string
    default: factory.commands

  consumerGroupId:
    description: Consumer group ID that Connectware uses when consuming
    type: string
    default: connectware

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

resources:
  # Producer connection to Confluent Cloud using SASL PLAIN over TLS
  confluentProducerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref bootstrapServer
        clientType: producer
        sasl:
          mechanism: plain
          username: !ref confluentApiKey
          password: !ref confluentApiSecret

  # Produces shop floor data to the Confluent Cloud topic
  machineDataEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref confluentProducerConnection
      write:
        topic: !ref produceTopic

  machineDataMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-data'
          publish:
            endpoint: !ref machineDataEndpoint
          rules:
            - transform:
                expression: '{ "value": [ { "value": $string($) } ] }'

  # Consumer connection to Confluent Cloud
  # A Kafka connection is either a producer or a consumer, never both
  confluentConsumerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref bootstrapServer
        clientType: consumer
        groupId: !ref consumerGroupId
        sasl:
          mechanism: plain
          username: !ref confluentApiKey
          password: !ref confluentApiSecret

  # Consumes the Confluent Cloud topic into the MQTT topic hierarchy
  commandsEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref confluentConsumerConnection
      subscribe:
        topic: !ref consumeTopic
        fromBeginning: false

  commandsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref commandsEndpoint
          publish:
            topic: !sub '${topicRoot}/confluent/commands'
          rules:
            - transform:
                expression: 'value.message'
