Validator Interface Update & Converter Changes

incubator/validator-avro.spec/src/main/scripts/io/aklivity/zilla/specs/validator/avro/config/validator.yaml

@@ -0,0 +1,50 @@
+#
+# Copyright 2021-2023 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+---
+name: test
+catalogs:
+  test0:
+    type: test
+    options:
+      schema: |
+        {
+          "fields": [
+            {
+              "name": "id",
+              "type": "string"
+            },
+            {
+              "name": "status",
+              "type": "string"
+            }
+          ],
+          "name": "Event",
+          "namespace": "io.aklivity.example",
+          "type": "record"
+        }
+bindings:
+  test:
+    kind: server
+    type: test
+    options:
+      value:
+        type: avro
+        expect: json
+        catalog:
+          catalog0:
+            - subject: test0
+              version: latest
+    exit: test

incubator/validator-avro.spec/src/main/scripts/io/aklivity/zilla/specs/validator/avro/schema/avro.schema.patch.json

@@ -3,5 +3,132 @@
         "op": "add",
         "path": "/$defs/validator/types/enum/-",
         "value": "avro"
+    },
+    {
+        "op": "add",
+        "path": "/$defs/validator/type/oneOf/-",
+        "value":
+        {
+            "if":
+            {
+                "properties":
+                {
+                    "type":
+                    {
+                        "const": "avro"
+                    }
+                }
+            },
+            "then":
+            {
+                "properties":
+                {
+                    "type":
+                    {
+                        "const": "avro"
+                    },
+                    "expect":
+                    {
+                        "type": "string",
+                        "enum":
+                        [
+                            "json"
+                        ]
+                    },
+                    "catalog":
+                    {
+                        "type": "object",
+                        "patternProperties":
+                        {
+                            "^[a-zA-Z]+[a-zA-Z0-9\\._\\-]*$":
+                            {
+                                "type": "array",
+                                "items":
+                                {
+                                    "oneOf":
+                                    [
+                                        {
+                                            "type": "object",
+                                            "properties":
+                                            {
+                                                "id":
+                                                {
+                                                    "type": "integer"
+                                                }
+                                            },
+                                            "required":
+                                            [
+                                                "id"
+                                            ],
+                                            "additionalProperties": false
+                                        },
+                                        {
+                                            "type": "object",
+                                            "properties":
+                                            {
+                                                "schema":
+                                                {
+                                                    "type": "string"
+                                                },
+                                                "version":
+                                                {
+                                                    "type": "string",
+                                                    "default": "latest"
+                                                }
+                                            },
+                                            "required":
+                                            [
+                                                "schema"
+                                            ],
+                                            "additionalProperties": false
+                                        },
+                                        {
+                                            "type": "object",
+                                            "properties":
+                                            {
+                                                "strategy":
+                                                {
+                                                    "type": "string"
+                                                },
+                                                "version":
+                                                {
+                                                    "type": "string",
+                                                    "default": "latest"
+                                                }
+                                            },
+                                            "required":
+                                            [
+                                                "strategy"
+                                            ],
+                                            "additionalProperties": false
+                                        },
+                                        {
+                                            "type": "object",
+                                            "properties":
+                                            {
+                                                "subject":
+                                                {
+                                                    "type": "string"
+                                                },
+                                                "version":
+                                                {
+                                                    "type": "string",
+                                                    "default": "latest"
+                                                }
+                                            },
+                                            "required":
+                                            [
+                                                "subject"
+                                            ],
+                                            "additionalProperties": false
+                                        }
+                                    ]
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
     }
 ]

incubator/validator-avro.spec/src/test/java/io/aklivity/zilla/specs/validator/avro/config/SchemaTest.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.specs.validator.avro.config;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+
+import jakarta.json.JsonObject;
+
+import org.junit.Rule;
+import org.junit.Test;
+
+import io.aklivity.zilla.specs.engine.config.ConfigSchemaRule;
+
+public class SchemaTest
+{
+    @Rule
+    public final ConfigSchemaRule schema = new ConfigSchemaRule()
+        .schemaPatch("io/aklivity/zilla/specs/engine/schema/binding/test.schema.patch.json")
+        .schemaPatch("io/aklivity/zilla/specs/engine/schema/catalog/test.schema.patch.json")
+        .schemaPatch("io/aklivity/zilla/specs/validator/avro/schema/avro.schema.patch.json")
+        .configurationRoot("io/aklivity/zilla/specs/validator/avro/config");
+
+    @Test
+    public void shouldValidateCatalog()
+    {
+        JsonObject config = schema.validate("validator.yaml");
+
+        assertThat(config, not(nullValue()));
+    }
+}

incubator/validator-avro/pom.xml

@@ -26,7 +26,7 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.80</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.92</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 

incubator/validator-avro/src/main/java/io/aklivity/zilla/runtime/validator/avro/AvroReadValueValidator.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.validator.avro;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.util.function.LongFunction;
+import java.util.function.ToLongFunction;
+
+import org.agrona.DirectBuffer;
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.JsonEncoder;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.apache.avro.specific.SpecificRecord;
+
+import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
+import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
+import io.aklivity.zilla.runtime.validator.avro.config.AvroValidatorConfig;
+
+public class AvroReadValueValidator extends AvroValueValidator
+{
+    public AvroReadValueValidator(
+        AvroValidatorConfig config,
+        ToLongFunction<String> resolveId,
+        LongFunction<CatalogHandler> supplyCatalog)
+    {
+        super(config, resolveId, supplyCatalog);
+    }
+
+    @Override
+    public int validate(
+        DirectBuffer data,
+        int index,
+        int length,
+        ValueConsumer next)
+    {
+        MutableDirectBuffer value = new UnsafeBuffer();
+        int valLength = -1;
+
+        byte[] payloadBytes = new byte[length];
+        data.getBytes(0, payloadBytes);
+        ByteBuffer byteBuf = ByteBuffer.wrap(payloadBytes);
+
+        int schemaId;
+        Schema schema;
+        if (byteBuf.get() == MAGIC_BYTE)
+        {
+            schemaId = byteBuf.getInt();
+            int size = length - 1 - 4;
+            byte[] valBytes = new byte[size];
+            data.getBytes(length - size, valBytes);
+            schema = fetchSchema(schemaId);
+            if (schema != null)
+            {
+                if ("json".equals(expect))
+                {
+                    byte[] record = serializeAvroRecord(schema, valBytes);
+                    value.wrap(record);
+                    valLength = record.length;
+                }
+                else if (validate(schema, valBytes))
+                {
+                    value.wrap(data);
+                    valLength = length;
+                }
+            }
+        }
+        else
+        {
+            schemaId = catalog != null ? catalog.id : 0;
+            schema = fetchSchema(schemaId);
+            if (schema != null)
+            {
+                if ("json".equals(expect))
+                {
+                    byte[] record = serializeAvroRecord(schema, payloadBytes);
+                    value.wrap(record);
+                    valLength = record.length;
+                }
+                else if (validate(schema, payloadBytes))
+                {
+                    value.wrap(data);
+                    valLength = length;
+                }
+            }
+        }
+
+        next.accept(value, index, valLength);
+        return valLength;
+    }
+
+    private byte[] serializeAvroRecord(
+        Schema schema,
+        byte[] payloadBytes)
+    {
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        try
+        {
+            reader = new GenericDatumReader(schema);
+            GenericRecord record = (GenericRecord) reader.read(null,
+                    decoder.binaryDecoder(payloadBytes, null));
+            JsonEncoder jsonEncoder = encoder.jsonEncoder(record.getSchema(), outputStream);
+            writer = record instanceof SpecificRecord ?
+                new SpecificDatumWriter<>(record.getSchema()) :
+                new GenericDatumWriter<>(record.getSchema());
+            writer.write(record, jsonEncoder);
+            jsonEncoder.flush();
+            outputStream.close();
+        }
+        catch (Exception e)
+        {
+        }
+        return outputStream.toByteArray();
+    }
+}