[WIP] Allow POJOs as CloudEvent data, round 3

api/src/main/java/io/cloudevents/CloudEvent.java

@@ -28,5 +28,5 @@ public interface CloudEvent extends CloudEventAttributes, CloudEventExtensions {
      * The event data
      */
     @Nullable
-    byte[] getData();
+    Object getData();
 }

api/src/main/java/io/cloudevents/rw/CloudEventRWException.java

@@ -25,6 +25,7 @@ public enum CloudEventRWExceptionKind {
         INVALID_ATTRIBUTE_TYPE,
         INVALID_ATTRIBUTE_VALUE,
         INVALID_EXTENSION_TYPE,
+        DATA_CODEC_ERROR,
         OTHER
     }
 
@@ -85,6 +86,14 @@ public static CloudEventRWException newInvalidExtensionType(String extensionName
         );
     }
 
+    public static CloudEventRWException newDataCodecError(Throwable cause) {
+        return new CloudEventRWException(
+            CloudEventRWExceptionKind.DATA_CODEC_ERROR,
+            "Error while serializing data",
+            cause
+        );
+    }
+
     public static CloudEventRWException newOther(Throwable cause) {
         return new CloudEventRWException(
             CloudEventRWExceptionKind.OTHER,

api/src/main/java/io/cloudevents/rw/CloudEventWriter.java

@@ -17,6 +17,8 @@
 
 package io.cloudevents.rw;
 
+import io.cloudevents.lang.Nullable;
+
 /**
  * Interface to write the content (CloudEvents attributes, extensions and payload) from a
  * {@link io.cloudevents.rw.CloudEventReader} to a new representation.
@@ -27,10 +29,13 @@ public interface CloudEventWriter<R> extends CloudEventAttributesWriter, CloudEv
 
     /**
      * End the visit with a data field
+     * <p>
+     * TODO add note that explains that contentType is already provided before visiting the attributes
+     * TODO explain that value cannot be a base64 serialized byte array
      *
      * @return an eventual return value
      */
-    R end(byte[] value) throws CloudEventRWException;
+    R end(@Nullable String contentType, Object value) throws CloudEventRWException;
 
     /**
      * End the visit

benchmarks/src/main/java/io/cloudevents/bench/jackson/JsonFormatDeserializationBenchmark.java

@@ -17,7 +17,7 @@
 
 package io.cloudevents.bench.jackson;
 
-import io.cloudevents.jackson.JsonFormat;
+import io.cloudevents.jackson.Json;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
 
@@ -31,7 +31,7 @@ public static class DeserializationState {
 
         public byte[] eventWithJson;
         public byte[] eventWithXml;
-        public JsonFormat format = new JsonFormat();
+        public Json format = new Json();
 
         public DeserializationState() {
             eventWithJson = format.serialize(V1_WITH_JSON_DATA_WITH_EXT);

benchmarks/src/main/java/io/cloudevents/bench/jackson/JsonFormatSerializationBenchmark.java

@@ -19,7 +19,7 @@
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
-import io.cloudevents.jackson.JsonFormat;
+import io.cloudevents.jackson.Json;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
 
@@ -32,7 +32,7 @@ public class JsonFormatSerializationBenchmark {
     public static class SerializationState {
         public CloudEvent eventWithJson = CloudEventBuilder.v1(V1_WITH_JSON_DATA_WITH_EXT).build();
         public CloudEvent eventWithXml = CloudEventBuilder.v1(V1_WITH_XML_DATA).build();
-        public JsonFormat format = new JsonFormat();
+        public Json format = new Json();
     }
 
     @Benchmark

benchmarks/src/main/java/io/cloudevents/bench/kafka/CloudEventToKafkaProducerMessageBenchmark.java

@@ -19,7 +19,7 @@
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
-import io.cloudevents.jackson.JsonFormat;
+import io.cloudevents.jackson.Json;
 import io.cloudevents.kafka.KafkaMessageFactory;
 import org.openjdk.jmh.annotations.*;
 import org.openjdk.jmh.infra.Blackhole;
@@ -50,7 +50,7 @@ public void testStructuredJsonEncoding(Event event, Blackhole bh) {
         bh.consume(
             KafkaMessageFactory
                 .createWriter("aaa")
-                .writeStructured(event.event, JsonFormat.CONTENT_TYPE)
+                .writeStructured(event.event, Json.CONTENT_TYPE)
         );
     }
 

benchmarks/src/main/java/io/cloudevents/bench/kafka/KafkaConsumerMessageToCloudEventBenchmark.java

@@ -17,7 +17,7 @@
 
 package io.cloudevents.bench.kafka;
 
-import io.cloudevents.jackson.JsonFormat;
+import io.cloudevents.jackson.Json;
 import io.cloudevents.kafka.KafkaMessageFactory;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -73,7 +73,7 @@ public StructuredJsonMessage() {
             // Hack to generate a consumer message
             ProducerRecord<Void, byte[]> inRecord = KafkaMessageFactory
                 .createWriter("aaa")
-                .writeStructured(V1_WITH_JSON_DATA_WITH_EXT, JsonFormat.CONTENT_TYPE);
+                .writeStructured(V1_WITH_JSON_DATA_WITH_EXT, Json.CONTENT_TYPE);
 
             this.message = new ConsumerRecord<>(
                 "aaa",

core/pom.xml

@@ -31,6 +31,7 @@
     <packaging>jar</packaging>
 
     <properties>
+        <jackson.version>2.11.2</jackson.version>
         <module-name>io.cloudevents.core</module-name>
     </properties>
 

core/src/main/java/io/cloudevents/core/builder/CloudEventBuilder.java

@@ -95,7 +95,7 @@ public interface CloudEventBuilder extends CloudEventWriter<CloudEvent> {
      * @param data data of the event
      * @return self
      */
-    CloudEventBuilder withData(byte[] data);
+    CloudEventBuilder withData(Object data);
 
     /**
      * Set the {@code datacontenttype} and {@code data} of the event
@@ -104,7 +104,7 @@ public interface CloudEventBuilder extends CloudEventWriter<CloudEvent> {
      * @param data            data of the event
      * @return self
      */
-    CloudEventBuilder withData(String dataContentType, byte[] data);
+    CloudEventBuilder withData(String dataContentType, Object data);
 
     /**
      * Set the {@code datacontenttype}, {@code dataschema} and {@code data} of the event
@@ -114,7 +114,7 @@ public interface CloudEventBuilder extends CloudEventWriter<CloudEvent> {
      * @param data            data of the event
      * @return self
      */
-    CloudEventBuilder withData(String dataContentType, URI dataSchema, byte[] data);
+    CloudEventBuilder withData(String dataContentType, URI dataSchema, Object data);
 
     /**
      * Set an extension with provided key and string value
@@ -214,4 +214,22 @@ static CloudEventBuilder fromSpecVersion(@Nonnull SpecVersion version) {
         );
     }
 
+    /**
+     * Create a new builder with a copy of the content inside the provided {@code event}.
+     *
+     * @param event event to copy
+     * @return a new builder
+     */
+    static CloudEventBuilder from(@Nonnull CloudEvent event) {
+        switch (event.getSpecVersion()) {
+            case V1:
+                return CloudEventBuilder.v1(event);
+            case V03:
+                return CloudEventBuilder.v03(event);
+        }
+        throw new IllegalStateException(
+            "The provided spec version doesn't exist. Please make sure your io.cloudevents deps versions are aligned."
+        );
+    }
+
 }

core/src/main/java/io/cloudevents/core/codec/DataSerializationException.java

@@ -0,0 +1,29 @@
+/*
+ * 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.core.codec;
+
+import io.cloudevents.core.format.EventFormat;
+
+/**
+ * Exception representing a serialization error while using an {@link EventFormat}.
+ */
+public class DataSerializationException extends RuntimeException {
+    public DataSerializationException(Throwable e) {
+        super(e);
+    }
+}

core/src/main/java/io/cloudevents/core/codec/EventDataCodec.java

@@ -0,0 +1,31 @@
+package io.cloudevents.core.codec;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventSerializationException;
+import io.cloudevents.lang.Nullable;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+/**
+ * This interface represents how to encode an event data to bytes, in order to write the payload on the wire.
+ */
+@ParametersAreNonnullByDefault
+public interface EventDataCodec {
+
+    /**
+     * Serialize a {@link CloudEvent} to a byte array.
+     *
+     * @param data the data to serialize.
+     * @return the byte representation of the provided event.
+     * @throws EventSerializationException if something goes wrong during serialization.
+     * @throws IllegalArgumentException    if this codec cannot serialize the provided data
+     */
+    byte[] serialize(@Nullable String dataContentType, Object data) throws DataSerializationException, IllegalArgumentException;
+
+    /**
+     * @param dataContentType the eventual content type to serialize
+     * @return true if this event codec instance can serialize the provided content type.
+     */
+    boolean canSerialize(@Nullable String dataContentType);
+
+}

core/src/main/java/io/cloudevents/core/codec/impl/TextEventDataCodec.java

@@ -0,0 +1,20 @@
+package io.cloudevents.core.codec.impl;
+
+import io.cloudevents.core.codec.DataSerializationException;
+import io.cloudevents.core.codec.EventDataCodec;
+
+public class TextEventDataCodec implements EventDataCodec {
+    @Override
+    public byte[] serialize(String dataContentType, Object data) throws DataSerializationException, IllegalArgumentException {
+        try {
+            return data.toString().getBytes();
+        } catch (Throwable e) {
+            throw new DataSerializationException(e);
+        }
+    }
+
+    @Override
+    public boolean canSerialize(String dataContentType) {
+        return "text/plain".equals(dataContentType);
+    }
+}

core/src/main/java/io/cloudevents/core/impl/BaseCloudEvent.java

@@ -26,16 +26,16 @@
 
 public abstract class BaseCloudEvent implements CloudEvent, CloudEventReader {
 
-    private final byte[] data;
+    private final Object data;
     protected final Map<String, Object> extensions;
 
-    protected BaseCloudEvent(byte[] data, Map<String, Object> extensions) {
+    protected BaseCloudEvent(Object data, Map<String, Object> extensions) {
         this.data = data;
         this.extensions = extensions != null ? extensions : new HashMap<>();
     }
 
     @Override
-    public byte[] getData() {
+    public Object getData() {
         return this.data;
     }
 
@@ -55,7 +55,7 @@ public <T extends CloudEventWriter<V>, V> V read(CloudEventWriterFactory<T, V> w
         this.readExtensions(visitor);
 
         if (this.data != null) {
-            return visitor.end(this.data);
+            return visitor.end(this.getDataContentType(), this.data);
         }
 
         return visitor.end();

core/src/main/java/io/cloudevents/core/impl/BaseCloudEventBuilder.java

@@ -27,12 +27,12 @@
 import java.util.HashMap;
 import java.util.Map;
 
-public abstract class BaseCloudEventBuilder<SELF extends BaseCloudEventBuilder<SELF, T>, T extends CloudEvent> implements CloudEventBuilder {
+public abstract class BaseCloudEventBuilder<SELF extends BaseCloudEventBuilder<SELF>> implements CloudEventBuilder {
 
     // This is a little trick for enabling fluency
     private final SELF self;
 
-    protected byte[] data;
+    protected Object data;
     protected Map<String, Object> extensions;
 
     @SuppressWarnings("unchecked")
@@ -55,21 +55,18 @@ public BaseCloudEventBuilder(CloudEvent event) {
 
     protected abstract void setAttributes(CloudEvent event);
 
-    //TODO builder should accept data as Object and use data codecs (that we need to implement)
-    // to encode data
-
-    public SELF withData(byte[] data) {
+    public SELF withData(Object data) {
         this.data = data;
         return this.self;
     }
 
-    public SELF withData(String dataContentType, byte[] data) {
+    public SELF withData(String dataContentType, Object data) {
         withDataContentType(dataContentType);
         withData(data);
         return this.self;
     }
 
-    public SELF withData(String dataContentType, URI dataSchema, byte[] data) {
+    public SELF withData(String dataContentType, URI dataSchema, Object data) {
         withDataContentType(dataContentType);
         withDataSchema(dataSchema);
         withData(data);
@@ -117,7 +114,7 @@ public void setExtension(String name, Boolean value) throws CloudEventRWExceptio
     }
 
     @Override
-    public CloudEvent end(byte[] value) throws CloudEventRWException {
+    public CloudEvent end(String contentType, Object value) throws CloudEventRWException {
         this.data = value;
         return build();
     }

core/src/main/java/io/cloudevents/core/impl/CloudEventReaderAdapter.java

@@ -22,7 +22,7 @@
 
 public class CloudEventReaderAdapter implements CloudEventReader {
 
-    private CloudEvent event;
+    private final CloudEvent event;
 
     CloudEventReaderAdapter(CloudEvent event) {
         this.event = event;
@@ -35,7 +35,7 @@ public <V extends CloudEventWriter<R>, R> R read(CloudEventWriterFactory<V, R> w
         this.readExtensions(visitor);
 
         if (event.getData() != null) {
-            return visitor.end(event.getData());
+            return visitor.end(event.getDataContentType(), event.getData());
         }
 
         return visitor.end();

core/src/main/java/io/cloudevents/core/impl/CloudEventUtils.java

@@ -26,7 +26,7 @@ private CloudEventUtils() {}
 
     /**
      * Convert a {@link CloudEvent} to a {@link CloudEventReader}. This method provides a default implementation
-     * for CloudEvent that doesn't implement CloudEventVisitable
+     * for {@link CloudEvent} implementations that don't implement {@link CloudEventReader}
      *
      * @param event the event to convert
      * @return the visitable implementation