support for Protobuf model

runtime/model-avro/src/main/java/io/aklivity/zilla/runtime/model/avro/internal/AvroField.java

@@ -0,0 +1,32 @@
+/*
+ * 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.model.avro.internal;
+
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+
+import io.aklivity.zilla.runtime.model.avro.internal.types.OctetsFW;
+
+public class AvroField
+{
+    public final OctetsFW value;
+    public final MutableDirectBuffer buffer;
+
+    public AvroField()
+    {
+        this.value = new OctetsFW();
+        this.buffer = new UnsafeBuffer(new byte[24]);
+    }
+}

runtime/model-avro/src/main/java/io/aklivity/zilla/runtime/model/avro/internal/AvroModelHandler.java

@@ -27,7 +27,6 @@
 import org.agrona.MutableDirectBuffer;
 import org.agrona.collections.Int2IntHashMap;
 import org.agrona.collections.Int2ObjectCache;
-import org.agrona.concurrent.UnsafeBuffer;
 import org.agrona.io.DirectBufferInputStream;
 import org.agrona.io.ExpandableDirectBufferOutputStream;
 import org.apache.avro.AvroRuntimeException;
@@ -65,7 +64,6 @@ public abstract class AvroModelHandler
     private static final int COMMA_LENGTH = ",".length();
     private static final int JSON_FIELD_ARRAY_LENGTH = "\"\":[]," .length() + COMMA_LENGTH * 100;
     private static final int JSON_FIELD_MAP_LENGTH = "\"\":{},".length();
-    private static final DirectBuffer EMPTY_BUFFER = new UnsafeBuffer();
 
     protected final SchemaConfig catalog;
     protected final CatalogHandler handler;
@@ -78,7 +76,7 @@ public abstract class AvroModelHandler
     protected final ExpandableDirectBufferOutputStream expandable;
     protected final DirectBufferInputStream in;
     protected final AvroModelEventContext event;
-    protected final Map<String, OctetsFW> extracted;
+    protected final Map<String, AvroField> extracted;
 
     private final Int2ObjectCache<Schema> schemas;
     private final Int2ObjectCache<GenericDatumReader<GenericRecord>> readers;
@@ -90,9 +88,8 @@ public abstract class AvroModelHandler
     private final AvroLongFW longRO;
     private final AvroFloatFW floatRO;
     private final AvroDoubleFW doubleRO;
-    private final MutableDirectBuffer text;
 
-    private int progress;
+    protected int progress;
 
     protected AvroModelHandler(
         AvroModelConfig config,
@@ -123,7 +120,6 @@ protected AvroModelHandler(
         this.longRO = new AvroLongFW();
         this.floatRO = new AvroFloatFW();
         this.doubleRO = new AvroDoubleFW();
-        this.text = new UnsafeBuffer(new byte[24]);
 
     }
 
@@ -136,10 +132,6 @@ protected final boolean validate(
         int length)
     {
         boolean status = false;
-        for (OctetsFW value: extracted.values())
-        {
-            value.wrap(EMPTY_BUFFER, 0, 0);
-        }
         try
         {
             GenericRecord record = supplyRecord(schemaId);
@@ -163,7 +155,7 @@ protected final boolean validate(
         return status;
     }
 
-    private void extractFields(
+    protected void extractFields(
         DirectBuffer buffer,
         int length,
         Schema schema)
@@ -304,7 +296,7 @@ private void extract(
         Schema schema,
         DirectBuffer data,
         int limit,
-        OctetsFW octets)
+        AvroField field)
     {
         switch (schema.getType())
         {
@@ -316,19 +308,22 @@ private void extract(
             AvroBytesFW bytes = bytesRO.wrap(data, progress, limit);
             OctetsFW value = bytes.value();
             progress = bytes.limit();
-            if (octets != null)
+            if (field != null)
             {
+                OctetsFW octets = field.value;
                 octets.wrap(value.buffer(), value.offset(), value.limit());
             }
             break;
+        case ENUM:
         case INT:
             AvroIntFW int32 = intRO.wrap(data, progress, limit);
             int intValue = int32.value();
             progress = int32.limit();
-            int length = text.putIntAscii(0, intValue);
-            if (octets != null)
+            if (field != null)
             {
-                octets.wrap(text, 0, length);
+                MutableDirectBuffer text = field.buffer;
+                int length = text.putIntAscii(0, intValue);
+                field.value.wrap(text, 0, length);
             }
             break;
         case FLOAT:
@@ -337,20 +332,22 @@ private void extract(
             DirectBuffer buffer = avroFloat.value().value();
             float floatValue = Float.intBitsToFloat(decodeNumberBytes(len, buffer));
             progress = avroFloat.limit();
-            length = text.putStringWithoutLengthAscii(0, String.valueOf(floatValue));
-            if (octets != null)
+            if (field != null)
             {
-                octets.wrap(text, 0, length);
+                MutableDirectBuffer text = field.buffer;
+                int length = text.putStringWithoutLengthAscii(0, String.valueOf(floatValue));
+                field.value.wrap(text, 0, length);
             }
             break;
         case LONG:
             AvroLongFW avroLong = longRO.wrap(data, progress, limit);
             long longValue = avroLong.value();
             progress = avroLong.limit();
-            length = text.putLongAscii(0, longValue);
-            if (octets != null)
+            if (field != null)
             {
-                octets.wrap(text, 0, length);
+                MutableDirectBuffer text = field.buffer;
+                int length = text.putLongAscii(0, longValue);
+                field.value.wrap(text, 0, length);
             }
             break;
         case DOUBLE:
@@ -365,20 +362,29 @@ private void extract(
             double doubleValue = Double.longBitsToDouble((((long) decoded) & 0xffffffffL) |
                 (((long) decodeNumberBytes(len, buffer)) << 32));
             progress = avroDouble.limit();
-            length = text.putStringWithoutLengthAscii(0, String.valueOf(doubleValue));
-            if (octets != null)
+            if (field != null)
             {
-                octets.wrap(text, 0, length);
+                MutableDirectBuffer text = field.buffer;
+                int length = text.putStringWithoutLengthAscii(0, String.valueOf(doubleValue));
+                field.value.wrap(text, 0, length);
             }
             break;
         case BOOLEAN:
             AvroBooleanFW avroBoolean = new AvroBooleanFW().wrap(data, progress, limit);
             value = avroBoolean.value();
             progress = avroBoolean.limit();
-            if (octets != null)
+            if (field != null)
             {
-                octets.wrap(value.buffer(), value.offset(), value.limit());
+                field.value.wrap(value.buffer(), value.offset(), value.limit());
+            }
+            break;
+        case FIXED:
+            int fixedSize = schema.getFixedSize();
+            if (field != null)
+            {
+                field.value.wrap(data, progress, progress + fixedSize);
             }
+            progress += fixedSize;
             break;
         }
     }

runtime/model-avro/src/main/java/io/aklivity/zilla/runtime/model/avro/internal/AvroReadConverterHandler.java

@@ -21,6 +21,7 @@
 import java.util.regex.Pattern;
 
 import org.agrona.DirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumReader;
@@ -38,6 +39,7 @@ public class AvroReadConverterHandler extends AvroModelHandler implements Conver
 {
     private static final String PATH = "^\\$\\.([A-Za-z_][A-Za-z0-9_]*)$";
     private static final Pattern PATH_PATTERN = Pattern.compile(PATH);
+    private static final DirectBuffer EMPTY_BUFFER = new UnsafeBuffer();
 
     private final Matcher matcher;
 
@@ -82,7 +84,7 @@ public void extract(
     {
         if (matcher.reset(path).matches())
         {
-            extracted.put(matcher.group(1), new OctetsFW());
+            extracted.put(matcher.group(1), new AvroField());
         }
     }
 
@@ -95,6 +97,10 @@ public int convert(
         int length,
         ValueConsumer next)
     {
+        for (AvroField field: extracted.values())
+        {
+            field.value.wrap(EMPTY_BUFFER, 0, 0);
+        }
         return handler.decode(traceId, bindingId, data, index, length, next, this::decodePayload);
     }
 
@@ -105,7 +111,7 @@ public int extractedLength(
         OctetsFW value = null;
         if (matcher.reset(path).matches())
         {
-            value = extracted.get(matcher.group(1));
+            value = extracted.get(matcher.group(1)).value;
         }
         return value != null ? value.sizeof() : 0;
     }
@@ -117,7 +123,7 @@ public void extracted(
     {
         if (matcher.reset(path).matches())
         {
-            OctetsFW value = extracted.get(matcher.group(1));
+            OctetsFW value = extracted.get(matcher.group(1)).value;
             if (value != null && value.sizeof() != 0)
             {
                 visitor.visit(value.buffer(), value.offset(), value.sizeof());
@@ -188,6 +194,9 @@ record = reader.read(record, decoderFactory.binaryDecoder(in, decoder));
                 JsonEncoder out = encoderFactory.jsonEncoder(schema, expandable);
                 writer.write(record, out);
                 out.flush();
+
+                progress = index;
+                extractFields(buffer, length, schema);
             }
         }
         catch (IOException | AvroRuntimeException ex)

runtime/model-avro/src/test/java/io/aklivity/zilla/runtime/model/avro/internal/AvroModelTest.java

@@ -286,10 +286,28 @@ public void shouldExtract()
         assertEquals(data.capacity(), converter.convert(0L, 0L, data, 0, data.capacity(), ValueConsumer.NOP));
 
         assertEquals(8, converter.extractedLength(stringPath));
-        final ConverterHandler.FieldVisitor visitor = (buffer, index, length) ->
+        ConverterHandler.FieldVisitor visitor = (buffer, index, length) ->
         {
             assertEquals("positive", buffer.getStringWithoutLengthUtf8(index, length));
         };
         converter.extracted(stringPath, visitor);
+
+        ConverterHandler.FieldVisitor doubleVisitor = (buffer, index, length) ->
+        {
+            assertEquals("1.2", buffer.getStringWithoutLengthUtf8(index, length));
+        };
+        converter.extracted(doublePath, doubleVisitor);
+
+        ConverterHandler.FieldVisitor intVisitor = (buffer, index, length) ->
+        {
+            assertEquals("1", buffer.getStringWithoutLengthUtf8(index, length));
+        };
+        converter.extracted(intPath, intVisitor);
+
+        ConverterHandler.FieldVisitor floatVisitor = (buffer, index, length) ->
+        {
+            assertEquals("2.2", buffer.getStringWithoutLengthUtf8(index, length));
+        };
+        converter.extracted(floatPath, floatVisitor);
     }
 }

runtime/model-protobuf/pom.xml

@@ -24,7 +24,7 @@
 </licenses>
 
 <properties>
-  <jacoco.coverage.ratio>0.92</jacoco.coverage.ratio>
+  <jacoco.coverage.ratio>0.89</jacoco.coverage.ratio>
   <jacoco.missed.count>0</jacoco.missed.count>
 </properties>
 

runtime/model-protobuf/src/main/java/io/aklivity/zilla/runtime/model/protobuf/internal/ProtobufField.java

@@ -0,0 +1,32 @@
+/*
+ * 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.model.protobuf.internal;
+
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
+
+import io.aklivity.zilla.runtime.model.protobuf.internal.types.OctetsFW;
+
+public class ProtobufField
+{
+    public final OctetsFW value;
+    public final MutableDirectBuffer buffer;
+
+    public ProtobufField()
+    {
+        this.value = new OctetsFW();
+        this.buffer = new UnsafeBuffer(new byte[24]);
+    }
+}