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

description: >
  Service commissioning file for the bidirectional integration between
  Connectware and a self-managed Apache Kafka cluster (Example)

metadata:
  name: Apache Kafka Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  brokers:
    description: Broker address of your Apache Kafka cluster
    type: string
    default: kafka-1.example.com:9093

  saslUsername:
    description: SASL username of the Connectware client
    type: string
    default: connectware

  saslPassword:
    description: SASL password of the Connectware client
    type: string

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

  consumeTopic:
    description: Kafka 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 using SASL SCRAM-SHA-512.
  # Configuring the sasl property makes the connector connect over TLS.
  # If your brokers use certificates from an internal CA, add the caCert
  # property with the Base64-encoded PEM content of the CA certificate.
  kafkaProducerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref brokers
        clientType: producer
        sasl:
          mechanism: scram-sha-512
          username: !ref saslUsername
          password: !ref saslPassword

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

  # Streams shop floor data from the MQTT topic hierarchy to Kafka.
  # The transform rule wraps each payload in the message format that the
  # Kafka connector expects.
  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 the same cluster.
  # A Kafka connection is either a producer or a consumer, never both.
  kafkaConsumerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref brokers
        clientType: consumer
        groupId: !ref consumerGroupId
        sasl:
          mechanism: scram-sha-512
          username: !ref saslUsername
          password: !ref saslPassword

  # Consumes new records from the Kafka topic
  commandsEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref kafkaConsumerConnection
      subscribe:
        topic: !ref consumeTopic
        fromBeginning: false

  # Publishes the body of every consumed record to an MQTT topic
  commandsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref commandsEndpoint
          publish:
            topic: !sub '${topicRoot}/kafka/commands'
          rules:
            - transform:
                expression: 'value.message'
