# ---------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ---------------------------------------------------------------------------
apiVersion: camel.apache.org/v1
kind: Kamelet
metadata:
  name: pulsar-source
  annotations:
    camel.apache.org/kamelet.support.level: "Stable"
    camel.apache.org/catalog.version: "4.10.2"
    camel.apache.org/kamelet.icon: "data:image/svg+xml;base64,[encoded data, 5640 characters]"
    camel.apache.org/provider: "Apache Software Foundation"
    camel.apache.org/kamelet.group: "Pulsar"
    camel.apache.org/kamelet.namespace: "Messaging"
  labels:
    camel.apache.org/kamelet.type: "source"
spec:
  dependencies:
    - "camel:pulsar"
    - "camel:kamelet"
    - "camel:core"
  definition:
    title: "Pulsar Source"
    description: "Receive data from Pulsar topics."
    required:
      - topicType
      - topic
      - namespaceName
      - tenant
      - serviceUrl
    properties:
      topic:
        title: Topic Name
        description: The topic name or regexp
        type: string
      tenant:
        title: Tenant Name
        description: The Tenant Name
        type: string
      topicType:
        title: Topic Type
        description: "The topic type."
        type: string
        enum: ["persistent", "non-persistent"]
      namespaceName:
        title: Pulsar Namespace Name
        description: The Pulsar Namespace Name
        type: string
      serviceUrl:
        title: Service URL
        description: The Pulsar Service URL to point while creating the client from URI.
        type: string        
      authenticationClass:
        title: Authentication Class
        description: The Authentication FQCN to be used while creating the client from URI.
        type: string
      authenticationParams:
        title: Authentication Params
        description: The Authentication Parameters to be used while creating the client from URI.
        type: string
      consumerNamePrefix:
        title: Consumer Name Prefix
        description: Prefix to add to consumer names when a SHARED or FAILOVER subscription is used
        type: string
        default: cons
      consumerQueueSize:
        title: Consumer Queue Size
        description: Size of the consumer queue
        type: integer
        default: 10
      deadLetterTopic:
        title: Dead Letter Topic
        description: "Name of the topic where the messages which fail maxRedeliverCount times will be sent. Note: if not set, default topic name will be topicName-subscriptionName-DLQ."
        type: integer
      maxRedeliverCount:
        title: Maximum Redelivery Count
        description: "Maximum number of times that a message will be redelivered before being sent to the dead letter queue. If this value is not set, no Dead Letter Policy will be created."
        type: integer
      negativeAckRedeliveryDelayMicros:
        title: Negative Ack Redelivery Delay in Microseconds
        description: "Set the negative acknowledgement delay."
        type: integer
        default: 60000000
      messageListener:
        title: Message Listener
        description: "Whether to use the messageListener interface, or to receive messages using a separate thread pool."
        type: boolean
        default: true
      numberOfConsumers:
        title: Number Of Consumers
        description: "Number of consumers."
        type: integer
        default: 1        
      numberOfConsumerThreads:
        title: Number Of Consumer Threads
        description: "Number of threads to receive and handle messages when using a separate thread pool."
        type: integer
        default: 1
      readCompacted:
        title: Read Compacted
        description: "Enable compacted topic reading."
        type: boolean
        default: false
      subscriptionInitialPosition:
        title: Subscription Initial Position
        description: "Control the initial position in the topic of a newly created subscription. Default is latest message."
        type: string
        default: LATEST
        enum: ["EARLIEST", "LATEST"]
      subscriptionName:
        title: Subscription Name
        description: "Name of the subscription to use."
        type: string
        default: subs
      subscriptionTopicsMode:
        title: Subscription Topics Mode
        description: "Determines to which topics this consumer should be subscribed to - Persistent, Non-Persistent, or both. Only used with pattern subscriptions."
        type: string
        default: PersistentOnly
        enum: ["PersistentOnly", "NonPersistentOnly", "AllTopics"]
      subscriptionType:
        title: Subscription Type
        description: "Type of the subscription."
        type: string
        default: EXCLUSIVE
        enum: ["EXCLUSIVE", "SHARED", "FAILOVER", "KEY_SHARED"]
      topicsPattern:
        title: Topic Pattern
        description: "Whether the topic is a pattern (regular expression) that allows the consumer to subscribe to all matching topics in the namespace."
        type: boolean
        default: false
    type: object
  template:
    from:
      uri: pulsar:{{topicType}}/{{tenant}}/{{namespaceName}}/{{topic}}
      parameters:
        serviceUrl: "{{serviceUrl}}"
        authenticationClass: "{{?authenticationClass}}"
        authenticationParams: "{{?authenticationParams}}"
        consumerNamePrefix: "{{?consumerNamePrefix}}"
        consumerQueueSize: "{{?consumerQueueSize}}"
        deadLetterTopic: "{{?deadLetterTopic}}"
        maxRedeliverCount: "{{?maxRedeliverCount}}"
        negativeAckRedeliveryDelayMicros: "{{?negativeAckRedeliveryDelayMicros}}"
        messageListener: "{{?messageListener}}"
        numberOfConsumers: "{{?numberOfConsumers}}"   
        numberOfConsumerThreads: "{{?numberOfConsumerThreads}}"
        readCompacted: "{{?readCompacted}}"
        subscriptionInitialPosition: "{{?subscriptionInitialPosition}}"
        subscriptionName: "{{?subscriptionName}}"
        subscriptionTopicsMode: "{{?subscriptionTopicsMode}}"
        subscriptionType: "{{?subscriptionType}}"
        topicsPattern: "{{?topicsPattern}}"
      steps:
        - to: "kamelet:sink"