openfunction_dapr_trigger

Follow this guide to install a Kafka server named kafka-server and a Topic named sample-topic.

Follow this guide to install DAPR.

Use kubectl to apply the manifests, then connect the container via nocalhost.

Create a DAPR Component:

apiVersion: dapr.io/v1alpha1kind: Componentmetadata:
name: kafka-bindingspec:
type: bindings.kafkaversion: v1metadata:
- name: topicsvalue: "sample-topic"
- name: brokersvalue: "kafka-server:9092"
- name: consumerGroupvalue: "group1"
- name: publishTopicvalue: "sample-topic"
- name: authRequiredvalue: "false"

FUNC_CONTEXT example:

{
"name": "ff-python",
"version": "v1",
"triggers": {
"dapr": [
{
"name": "kafka-binding",
"type": "bindings.kafka"
}
]
},
"port": 50055
}

After logging into the container terminal, export FUNC_CONTEXT:

export FUNC_CONTEXT='{"name":"ff-python","version":"v1","triggers":{"dapr":[{"name":"kafka-binding","type":"bindings.kafka"}]},"port":50055}'

Run function:

ff --source examples/openfunction_dapr_trigger/user_function.py --target user_function

Use the DAPR client to call function or you can apply the caller

importjsonimporttimefromdapr.clientsimportDaprClientwithDaprClient() asd:
n=0whileTrue:
n+=1req_data= {
'id': n,
'message': 'hello world'
}
print(f'Sending message id: {req_data["id"]}, message "{req_data["message"]}"', flush=True)
# Create a typed message with content type and body_=d.invoke_binding('kafka-binding', 'create', json.dumps(req_data))
time.sleep(2)