Avro format implementation

formats/avro/README.md

@@ -0,0 +1,4 @@
+# CloudEvents Avro Format
+
+This project provides functionality for the Java SDK to handle the
+[avro format](https://github.com/cloudevents/spec/blob/v1.0.1/avro-format.md).

formats/avro/pom.xml

@@ -0,0 +1,120 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2021-Present The CloudEvents Authors
+  ~ <p>
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ you may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~ <p>
+  ~ http://www.apache.org/licenses/LICENSE-2.0
+  ~ <p>
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.cloudevents</groupId>
+        <artifactId>cloudevents-parent</artifactId>
+        <version>2.3.0-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>cloudevents-avro</artifactId>
+    <name>CloudEvents - Avro</name>
+    <packaging>jar</packaging>
+
+    <properties>
+        <avro.version>1.9.2</avro.version>
+        <module-name>io.cloudevents.formats.avro</module-name>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.avro</groupId>
+                <artifactId>avro-maven-plugin</artifactId>
+                <version>1.10.2</version>
+                <executions>
+                    <execution>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>schema</goal>
+                        </goals>
+                        <configuration>
+                            <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
+                            <outputDirectory>${project.build.directory}/generated-sources/java/</outputDirectory>
+                            <stringType>String</stringType>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>build-helper-maven-plugin</artifactId>
+                <version>3.2.0</version>
+                <executions>
+                    <execution>
+                        <id>add-source</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>add-source</goal>
+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>${project.build.directory}/generated-sources/java/</source>
+                            </sources>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.cloudevents</groupId>
+            <artifactId>cloudevents-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>${avro.version}</version>
+        </dependency>
+
+        <!-- Test deps -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>${junit-jupiter.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj-core.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.cloudevents</groupId>
+            <artifactId>cloudevents-core</artifactId>
+            <classifier>tests</classifier>
+            <type>test-jar</type>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

formats/avro/src/main/avro/spec.avsc

@@ -0,0 +1,64 @@
+{
+  "namespace":"io.cloudevents",
+  "type":"record",
+  "name":"AvroCloudEvent",
+  "version":"1.0",
+  "doc":"Avro Event Format for CloudEvents",
+  "fields":[
+    {
+      "name":"attribute",
+      "type":{
+        "type":"map",
+        "values":[
+          "null",
+          "boolean",
+          "int",
+          "string",
+          "bytes"
+        ]
+      }
+    },
+    {
+      "name": "data",
+      "type": [
+        "bytes",
+        "null",
+        "boolean",
+        {
+          "type": "map",
+          "values": [
+            "null",
+            "boolean",
+            {
+              "type": "record",
+              "name": "AvroCloudEventData",
+              "doc": "Representation of a JSON Value",
+              "fields": [
+                {
+                  "name": "value",
+                  "type": {
+                    "type": "map",
+                    "values": [
+                      "null",
+                      "boolean",
+                      { "type": "map", "values": "AvroCloudEventData" },
+                      { "type": "array", "items": "AvroCloudEventData" },
+                      "double",
+                      "string"
+                    ]
+                  }
+                }
+              ]
+            },
+            "double",
+            "string"
+          ]
+        },
+        { "type": "array", "items": "AvroCloudEventData" },
+        "double",
+        "string"
+      ]
+    }
+  ]
+}
+

formats/avro/src/main/java/io/cloudevents/avro/AvroCloudEventDataWrapper.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2018-Present The CloudEvents Authors
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package io.cloudevents.avro;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.cloudevents.AvroCloudEventData;
+import io.cloudevents.CloudEventData;
+
+/**
+ * Encode JSON style cloudevent data into Avro format.
+ *
+ */
+public class AvroCloudEventDataWrapper implements CloudEventData {
+
+    private AvroCloudEventData avroCloudEventData;
+
+    /**
+     * Wraps a JSON object-like data structure.
+     */
+    public AvroCloudEventDataWrapper(Map<String, Object> data) {
+        avroCloudEventData = new AvroCloudEventData();
+        avroCloudEventData.setValue(data);
+    }
+
+    @Override
+    public byte[] toBytes() {
+        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+        try {
+            AvroCloudEventData.getEncoder().encode(this.avroCloudEventData, bytes);
+        } catch (IOException ignore) {
+            // ignored
+        }
+
+        return bytes.toByteArray();
+    }
+}

formats/avro/src/main/java/io/cloudevents/avro/AvroDeserializer.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2018-Present The CloudEvents Authors
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.cloudevents.avro;
+
+import java.util.Map;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.time.OffsetDateTime;
+
+import io.cloudevents.CloudEventData;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.AvroCloudEvent;
+import io.cloudevents.core.data.BytesCloudEventData;
+import io.cloudevents.core.v1.CloudEventV1;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventReader;
+import io.cloudevents.rw.CloudEventDataMapper;
+import io.cloudevents.rw.CloudEventWriter;
+import io.cloudevents.rw.CloudEventWriterFactory;
+
+class AvroDeserializer implements CloudEventReader {
+
+    private final AvroCloudEvent avroCloudEvent;
+
+    public AvroDeserializer(AvroCloudEvent avroCloudEvent) {
+        this.avroCloudEvent = avroCloudEvent;
+    }
+
+    @Override
+    public <W extends CloudEventWriter<R>, R> R read(CloudEventWriterFactory<W, R> writerFactory,
+                                                     CloudEventDataMapper<? extends CloudEventData> mapper) throws CloudEventRWException {
+        Map<String, Object> avroCloudEventAttrs = this.avroCloudEvent.getAttribute();
+        SpecVersion specVersion = SpecVersion.parse((String)avroCloudEventAttrs.get(CloudEventV1.SPECVERSION));
+        final CloudEventWriter<R> writer = writerFactory.create(specVersion);
+
+        for (Map.Entry<String, Object> entry: avroCloudEventAttrs.entrySet()) {
+            String key = entry.getKey().toString();
+
+            switch(key) {
+                case CloudEventV1.SPECVERSION:
+                    continue;
+                case CloudEventV1.TIME: {
+                    // OffsetDateTime
+                    OffsetDateTime value = OffsetDateTime.parse((String) entry.getValue());
+                    writer.withContextAttribute(key, value);
+                };
+                case CloudEventV1.DATASCHEMA: {
+                    // URI
+                    URI value = URI.create((String) entry.getValue());
+                    writer.withContextAttribute(key, value);
+                };
+                default:
+                    writer.withContextAttribute(key, (String) entry.getValue());
+            }
+        }
+
+        ByteBuffer buffer = (ByteBuffer) this.avroCloudEvent.getData();
+
+        if (buffer != null) {
+            byte[] data = new byte[buffer.remaining()];
+            buffer.get(data);
+            return writer.end(mapper.map(BytesCloudEventData.wrap(data)));
+        } else {
+            return writer.end();
+        }
+    }
+
+}

formats/avro/src/main/java/io/cloudevents/avro/AvroFormat.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2018-Present The CloudEvents Authors
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.cloudevents.avro;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.CloudEventData;
+import io.cloudevents.AvroCloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.cloudevents.core.format.EventDeserializationException;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.format.EventSerializationException;
+import io.cloudevents.rw.CloudEventDataMapper;
+
+public class AvroFormat implements EventFormat {
+
+    public static final String AVRO_CONTENT_TYPE = "application/avro";
+
+    @Override
+    public byte[] serialize(CloudEvent event) throws EventSerializationException {
+        AvroCloudEvent avroCloudEvent = AvroSerializer.toAvro(event);
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+        try {
+            AvroCloudEvent.getEncoder().encode(avroCloudEvent, output);
+        } catch (IOException e) {
+            throw new EventSerializationException(e);
+        }
+
+        return output.toByteArray();
+    }
+
+    @Override
+    public CloudEvent deserialize(byte[] bytes, CloudEventDataMapper<? extends CloudEventData> mapper)
+        throws EventDeserializationException {
+        ByteArrayInputStream input = new ByteArrayInputStream(bytes);
+
+         try {
+             AvroCloudEvent avroCloudEvent = AvroCloudEvent.getDecoder().decode(input);
+
+             return new AvroDeserializer(avroCloudEvent).read(CloudEventBuilder::fromSpecVersion, mapper);
+         } catch (IOException e) {
+             throw new EventDeserializationException(e);
+         }
+    }
+
+    @Override
+    public String serializedContentType() {
+        return AVRO_CONTENT_TYPE;
+    }
+}