matzew
commented
2 years ago I use a "processor", like:
package functions;
import io.quarkus.funqy.Funq;
import io.quarkus.funqy.knative.events.CloudEvent;
import io.quarkus.funqy.knative.events.CloudEventBuilder;
import java.util.UUID;
public class Function {
@Funq
public CloudEvent<Output> function(CloudEvent<Input> input) {
// work with the received Cloud event:
// do some sort of processing with it...
System.out.println(" We got: " + input);
// create some other business object:
Output output = new Output("HEllo");
// return the object to the broker,
// wrapped as a cloud event:
CloudEvent<Output> response = CloudEventBuilder.create()
.source("/service/processors/1/random")
.type("io.openshift.smart.event.processor")
.id(UUID.randomUUID().toString())
.build(output);
System.out.println("We sent back: " + response);
return response;
}
}and a receiver (the canonical event-display).
Here is the manifests:
apiVersion: v1
kind: Service
metadata:
name: processor
spec:
selector:
app: processor
ports:
- port: 80
protocol: TCP
targetPort: processor
name: http
---
apiVersion: v1
kind: Pod
metadata:
name: processor
labels:
app: processor
spec:
containers:
- name: processor
image: docker.io/matzew/processor-demo:latest
imagePullPolicy: Always
ports:
- containerPort: 8080
protocol: TCP
name: processor
---
apiVersion: v1
kind: Service
metadata:
name: receiver
spec:
selector:
app: receiver
ports:
- port: 80
protocol: TCP
targetPort: receiver
name: http
---
apiVersion: v1
kind: Pod
metadata:
name: receiver
labels:
app: receiver
spec:
containers:
- name: receiver
image: quay.io/openshift-knative/knative-eventing-sources-event-display
imagePullPolicy: Always
ports:
- containerPort: 8080
protocol: TCP
name: receiver
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: processor
spec:
broker: my-kafkachannel-advanced-broker
filter:
attributes:
smartevent: super-duper-event-extension
subscriber:
ref:
apiVersion: v1
kind: Service
name: processor
---
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: receiver
spec:
broker: my-kafkachannel-advanced-broker
filter:
attributes:
type: io.openshift.smart.event.processor
subscriber:
ref:
apiVersion: v1
kind: Service
name: receiver
---for the broker I use this config: https://github.com/openshift-knative/eventing-hyperfoil-benchmark/blob/main/testcases/broker/eventing-broker/broker-kc-advanced-config/100-broker.yaml
aliok
pierDipi
Using the channel in the MT Broker, I get this dispatcher log: