fix time serde and add data deserialization support

Signed-off-by: Ning Sun <[email protected]>

Author: sunng87

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

@@ -19,16 +19,12 @@
 
 import java.util.Map;
 import java.net.URI;
-import java.time.Instant;
 import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
 
-import io.cloudevents.CloudEvent;
 import io.cloudevents.CloudEventData;
 import io.cloudevents.SpecVersion;
 import io.cloudevents.AvroCloudEvent;
-import io.cloudevents.AvroCloudEventData;
-import io.cloudevents.core.builder.CloudEventBuilder;
+import io.cloudevents.core.data.BytesCloudEventData;
 import io.cloudevents.core.v1.CloudEventV1;
 import io.cloudevents.rw.CloudEventRWException;
 import io.cloudevents.rw.CloudEventReader;
@@ -57,9 +53,7 @@ public <W extends CloudEventWriter<R>, R> R read(CloudEventWriterFactory<W, R> w
 
             if (key.equals(CloudEventV1.TIME)) {
                 // OffsetDateTime
-                Long timeAsLong = (Long) entry.getValue();
-                Instant timeAsInstant = Instant.ofEpochMilli(timeAsLong);
-                OffsetDateTime value = OffsetDateTime.ofInstant(timeAsInstant, ZoneOffset.UTC);
+                OffsetDateTime value = OffsetDateTime.parse((String) entry.getValue());
                 writer.withContextAttribute(key, value);
 
             } else if (key.equals(CloudEventV1.DATASCHEMA)) {
@@ -72,8 +66,13 @@ public <W extends CloudEventWriter<R>, R> R read(CloudEventWriterFactory<W, R> w
             }
         }
 
-        // TOOD: data
-        return writer.end();
+        byte[] data = (byte[]) this.avroCloudEvent.getData();
+
+        if (data != null) {
+            return writer.end(mapper.map(BytesCloudEventData.wrap(data)));
+        } else {
+            return writer.end();
+        }
     }
 
 }

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

@@ -24,6 +24,7 @@
 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;
@@ -55,7 +56,7 @@ public CloudEvent deserialize(byte[] bytes, CloudEventDataMapper<? extends Cloud
          try {
              AvroCloudEvent avroCloudEvent = AvroCloudEvent.getDecoder().decode(input);
 
-             return AvroDeserializer.fromAvro(avroCloudEvent);
+             return new AvroDeserializer(avroCloudEvent).read(CloudEventBuilder::fromSpecVersion, mapper);
          } catch (IOException e) {
              throw new EventDeserializationException(e);
          }

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

@@ -37,8 +37,8 @@ public static final AvroCloudEvent toAvro(CloudEvent e) {
         attrs.put(CloudEventV1.SPECVERSION, e.getSpecVersion().toString());
         attrs.put(CloudEventV1.ID, e.getId());
         attrs.put(CloudEventV1.SOURCE, e.getSource());
-        // convert to long
-        attrs.put(CloudEventV1.TIME, e.getTime().toInstant().toEpochMilli());
+        // convert to string
+        attrs.put(CloudEventV1.TIME, e.getTime().toString());
         // convert
         attrs.put(CloudEventV1.DATASCHEMA, e.getDataSchema().toString());
         attrs.put(CloudEventV1.SUBJECT, e.getSubject());