Extend Expression Language to Access HTTP Headers and to Use Multiple Parameters

examples/http.kafka.avro.json/README.md

@@ -14,7 +14,23 @@ To `start` the Docker Compose stack defined in the [compose.yaml](compose.yaml)
 docker compose up -d
 ```
 
-### Register Schema
+### Register Schemas
+
+```bash
+curl 'http://localhost:8081/subjects/items-snapshots-key/versions' \
+--header 'Content-Type: application/json' \
+--data '{
+  "schema":
+    "{\"fields\":[{\"name\":\"item_id\",\"type\":\"string\"},{\"name\":\"user_id\",\"type\":\"string\"}],\"name\":\"EventKey\",\"namespace\":\"io.aklivity.example\",\"type\":\"record\"}",
+  "schemaType": "AVRO"
+}'
+```
+
+output:
+
+```text
+{"id":1}%
+```
 
 ```bash
 curl 'http://localhost:8081/subjects/items-snapshots-value/versions' \
@@ -29,16 +45,18 @@ curl 'http://localhost:8081/subjects/items-snapshots-value/versions' \
 output:
 
 ```text
-{"id":1}%
+{"id":2}%
 ```
 
 ## Validate created Schema
 
 ```bash
 curl 'http://localhost:8081/schemas/ids/1'
+curl 'http://localhost:8081/schemas/ids/2'
 ```
 
 ```bash
+curl 'http://localhost:8081/subjects/items-snapshots-key/versions/latest'
 curl 'http://localhost:8081/subjects/items-snapshots-value/versions/latest'
 ```
 
@@ -47,7 +65,7 @@ curl 'http://localhost:8081/subjects/items-snapshots-value/versions/latest'
 `POST` request
 
 ```bash
-curl -k -v http://localhost:7114/items -H 'Idempotency-Key: 1'  -H 'Content-Type: application/json' -d '{"id": "123","status": "OK"}'
+curl -k -v http://localhost:7114/items -H 'X-User-ID: user1' -H 'Idempotency-Key: 1'  -H 'Content-Type: application/json' -d '{"id": "123","status": "OK"}'
 ```
 
 output:
@@ -69,7 +87,7 @@ output:
 `GET` request to fetch specific item.
 
 ```bash
-curl -k http://localhost:7114/items/1
+curl -k http://localhost:7114/items/1 -H 'X-User-ID: user1'
 ```
 
 output:
@@ -90,7 +108,7 @@ output:
 `POST` request.
 
 ```bash
-curl -k -v http://localhost:7114/items -H 'Idempotency-Key: 2'  -H 'Content-Type: application/json' -d '{"id": 123,"status": "OK"}'
+curl -k -v http://localhost:7114/items -H 'X-User-ID: user1' -H 'Idempotency-Key: 2'  -H 'Content-Type: application/json' -d '{"id": 123,"status": "OK"}'
 ```
 
 output:
@@ -112,7 +130,7 @@ output:
 `GET` request to verify whether Invalid event is produced
 
 ```bash
-curl -k -v http://localhost:7114/items/2
+curl -k -v http://localhost:7114/items/2 -H 'X-User-ID: user1'
 ```
 
 output:

examples/http.kafka.avro.json/compose.yaml

@@ -17,7 +17,7 @@ services:
       ZILLA_INCUBATOR_ENABLED: "true"
     volumes:
       - ./etc:/etc/zilla
-    command: start -v -e
+    command: start -v -e --config /etc/zilla/zilla.yaml
 
   kafka:
     image: bitnami/kafka:3.5
@@ -78,6 +78,7 @@ services:
     environment:
       KAFKA_CLUSTERS_0_NAME: local
       KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka.examples.dev:29092
+      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://karapace-registry:8081
 
   kafkacat:
     image: confluentinc/cp-kafkacat:7.1.9

examples/http.kafka.avro.json/etc/zilla.yaml

@@ -37,23 +37,41 @@ bindings:
         with:
           capability: produce
           topic: items-snapshots
-          key: ${idempotencyKey}
+          key:  |
+            { 
+              "item_id": "${idempotencyKey}",
+              "user_id": "${headers.x-user-id}"
+            }
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: PUT
             path: /items/{id}
         exit: north_kafka_cache_client
         with:
           capability: produce
           topic: items-snapshots
-          key: ${params.id}
+          key:  |
+            { 
+              "item_id": "${idempotencyKey}",
+              "user_id": "${headers.x-user-id}"
+            }
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: DELETE
             path: /items/{id}
         exit: north_kafka_cache_client
         with:
           capability: produce
           topic: items-snapshots
-          key: ${params.id}
+          key: |
+            { 
+              "item_id": "${idempotencyKey}",
+              "user_id": "${headers.x-user-id}"
+            }
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: GET
             path: /items
@@ -63,6 +81,9 @@ bindings:
           topic: items-snapshots
           merge:
             content-type: application/json
+          filters:
+            - headers:
+                user_id: ${headers.x-user-id}
       - when:
           - method: GET
             path: /items/{id}
@@ -71,13 +92,22 @@ bindings:
           capability: fetch
           topic: items-snapshots
           filters:
-            - key: ${params.id}
+            # The flatten key is extracted from the original key object in the cache server
+            # for easier comparison.
+            - key: ${params.id}.${headers.x-user-id}
   north_kafka_cache_client:
     type: kafka
     kind: cache_client
     options:
       topics:
         - name: items-snapshots
+          key:
+            model: avro
+            view: json
+            catalog:
+              my_catalog:
+                - strategy: topic
+                  version: latest
           value:
             model: avro
             view: json
@@ -94,13 +124,22 @@ bindings:
         - items-snapshots
       topics:
         - name: items-snapshots
+          key:
+            model: avro
+            view: json
+            catalog:
+              my_catalog:
+                - strategy: topic
+                  version: latest
           value:
             model: avro
             view: json
             catalog:
               my_catalog:
                 - strategy: topic
                   version: latest
+          transforms:
+            - extract-key: ${message.key.item_id}.${message.key.user_id}
     exit: south_kafka_client
   south_kafka_client:
     type: kafka

examples/http.kafka.crud/README.md

@@ -32,7 +32,7 @@ docker compose up -d
 Note: You can remove `-H 'Idempotency-Key: 1'` to generate random key.
 
 ```bash
-curl -k -v -X POST http://localhost:7114/items -H 'Idempotency-Key: 1'  -H 'Content-Type: application/json' -d '{"greeting":"Hello, world1"}'
+curl -k -v -X POST http://localhost:7114/items -H 'x-user-id: user1' -H 'Idempotency-Key: 1'  -H 'Content-Type: application/json' -d '{"greeting":"Hello, world1"}'
 ```
 
 output:
@@ -54,7 +54,7 @@ HTTP/2 204
 `GET` request to fetch specific item.
 
 ```bash
-curl -k -v http://localhost:7114/items/1
+curl -k -v http://localhost:7114/items/1 -H 'x-user-id: user1'
 ```
 
 output:
@@ -74,7 +74,7 @@ output:
 `PUT` request to update specific item.
 
 ```bash
-curl -k -v -X PUT http://localhost:7114/items/1 -H 'Content-Type: application/json' -d '{"greeting":"Hello, world2"}'
+curl -k -v -X PUT http://localhost:7114/items/1 -H 'x-user-id: user1' -H 'Content-Type: application/json' -d '{"greeting":"Hello, world2"}'
 ```
 
 output:
@@ -96,7 +96,7 @@ HTTP/2 204
 `DELETE` request to delete specific item.
 
 ```bash
-curl -k -v -X DELETE http://localhost:7114/items/1
+curl -k -v -X DELETE http://localhost:7114/items/1 -H 'x-user-id: user1'
 ```
 
 output:

examples/http.kafka.crud/etc/zilla.yaml

@@ -31,23 +31,29 @@ bindings:
         with:
           capability: produce
           topic: items-snapshots
-          key: ${idempotencyKey}
+          key: ${idempotencyKey}.${headers.x-user-id}
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: PUT
             path: /items/{id}
         exit: north_kafka_cache_client
         with:
           capability: produce
           topic: items-snapshots
-          key: ${params.id}
+          key: ${params.id}.${headers.x-user-id}
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: DELETE
             path: /items/{id}
         exit: north_kafka_cache_client
         with:
           capability: produce
           topic: items-snapshots
-          key: ${params.id}
+          key: ${params.id}.${headers.x-user-id}
+          overrides:
+            user_id: ${headers.x-user-id}
       - when:
           - method: GET
             path: /items
@@ -57,6 +63,9 @@ bindings:
           topic: items-snapshots
           merge:
             content-type: application/json
+          filters:
+            - headers:
+                user_id: ${headers.x-user-id}
       - when:
           - method: GET
             path: /items/{id}
@@ -65,7 +74,7 @@ bindings:
           capability: fetch
           topic: items-snapshots
           filters:
-            - key: ${params.id}
+            - key: ${params.id}.${headers.x-user-id}
   north_kafka_cache_client:
     type: kafka
     kind: cache_client