---
title: Add Event Source and Trigger
type: proposal
menu: proposals
status: accepted
owner: benjaminhuo
---

## Summary

## Motivation

## Goals

## Proposal

### Argo-Events EventSource and Trigger

apiVersion: core.openfunction.io/v1alpha1
kind: EventSource
metadata:
  name: kafka
spec:
  kafka:
    example:
      # kafka broker url
      url: kafka.argo-events:9092
      # name of the kafka topic
      topic: topic-2
      # jsonBody specifies that all event body payload coming from this
      # source will be JSON
      jsonBody: true
      # partition id
      # optional backoff time for connection retries.
      # if not provided, default connection backoff time will be used.
      connectionBackoff:
        # duration in nanoseconds, or strings like "3s", "2m". following value is 10 seconds
        duration: 10s
        # how many backoffs
        steps: 5
        # factor to increase on each step.
        # setting factor > 1 makes backoff exponential.
        factor: 2
        jitter: 0.2
      consumerGroup:
        groupName: test-group
        oldest: false
        rebalanceStrategy: range
  sink:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: event-display
      namespace: default
---
apiVersion: core.openfunction.io/v1alpha1
kind: EventSource
metadata:
  name: mqtt
spec:
  mqtt:
    example:
      # mqtt broker url
      url: tcp://mqtt.argo-events:1883
      # name of the popic
      topic: bar
      # jsonBody specifies that all event body payload coming from this
      # source will be JSON
      jsonBody: true
      # client id
      clientId: "2345"
      # optional backoff time for connection retries.
      # if not provided, default connection backoff time will be used.
      connectionBackoff:
        # duration in nanoseconds, or strings like "2s". following value is 10 seconds
        duration: 10s
        # how many backoffs
        steps: 5
        # factor to increase on each step.
        # setting factor > 1 makes backoff exponential.
        factor: 2
        jitter: 0.2
  sink:
    ref:
      apiVersion: dapr.io/v1alpha1
      kind: Component
      name: dapr-pubsub-kafka # or dapr-pubsub-natss
      namespace: default 
---
apiVersion: core.openfunction.io/v1alpha1
kind: Trigger
metadata:
  name: func1
spec:
  eventSources:
    - name: mqtt
      eventSourceName: mqtt
      eventName: example
      filters:
        data:
          - path: body.value
            type: number
            comparator: ">"
            value:
              - "50.0"
    - name: kafka
      eventSourceName: kafka
      eventName: example
      filters:
        data:
          - path: bucket
            type: string
            value:
              - func1-input1
              - func1-input2
  subscribers:
    - condition: kafka
      ref:
        apiVersion: serving.knative.dev/v1
        kind: Service
        name: knative-func1
        namespace: default   
    - condition: mqtt
      ref:
        apiVersion: dapr.io/v1alpha1
        kind: Component
        name: func2-kafka-input-binding
        namespace: default
    - condition: mqtt || kafka
      ref:
        apiVersion: dapr.io/v1alpha1
        kind: Component
        name: func3-kafka-input-binding
        namespace: default   

### Dapr Kafka binding EventSource, Sink and Trigger

---          
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: func1-kafka-input-binding
  namespace: default
spec:
  type: bindings.kafka
  version: v1
  metadata:
  - name: topics # Optional. in use for input bindings
    value: inputTopic1,inputTopic2
  - name: brokers
    value: localhost:9092,localhost:9093
  - name: consumerGroup
    value: group1
  - name: publishTopic # Optional. in use for output bindings
    value: outputTopic
  - name: authRequired # Required. default: "true"
    value: "false"
  - name: saslUsername # Optional.
    value: "user"
  - name: saslPassword # Optional.
    value: "password"
  - name: maxMessageBytes # Optional.
    value: 1024 

### Dapr Kafka pub/sub EventSource

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: dapr-pubsub-natss
  namespace: default
spec:
  type: pubsub.natsstreaming
  version: v1
  metadata:
  - name: natsURL
    value: "nats://localhost:4222"
  - name: natsStreamingClusterID
    value: "clusterId"
    # below are subscription configuration.
  - name: subscriptionType
    value: <REPLACE-WITH-SUBSCRIPTION-TYPE> # Required. Allowed values: topic, queue.
  - name: ackWaitTime
    value: "" # Optional.
  - name: maxInFlight
    value: "" # Optional.
  - name: durableSubscriptionName
    value: "" # Optional.
  # following subscription options - only one can be used
  - name: deliverNew
    value: <bool>
  - name: startAtSequence
    value: 1
  - name: startWithLastReceived
    value: false
  - name: deliverAll
    value: false
  - name: startAtTimeDelta
    value: ""
  - name: startAtTime
    value: ""
  - name: startAtTimeFormat
    value: ""