wip improve serializer and deserializer

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

Author: sunng87

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

@@ -17,16 +17,63 @@
 
 package io.cloudevents.avro;
 
+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.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;
+
+public 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<CharSequence, Object> avroCloudEventAttrs = this.avroCloudEvent.getAttribute();
+        SpecVersion specVersion = SpecVersion.parse((String)avroCloudEventAttrs.get(CloudEventV1.SPECVERSION));
+        final CloudEventWriter<R> writer = writerFactory.create(specVersion);
+
+        for (Map.Entry<CharSequence, Object> entry: avroCloudEventAttrs.entrySet()) {
+            String key = entry.getKey().toString();
 
-public class AvroDeserializer {
+            if (key.equals(CloudEventV1.TIME)) {
+                // OffsetDateTime
+                Long timeAsLong = (Long) entry.getValue();
+                Instant timeAsInstant = Instant.ofEpochMilli(timeAsLong);
+                OffsetDateTime value = OffsetDateTime.ofInstant(timeAsInstant, ZoneOffset.UTC);
+                writer.withContextAttribute(key, value);
 
-    public static CloudEvent fromAvro(AvroCloudEvent avroCloudEvent) {
+            } else if (key.equals(CloudEventV1.DATASCHEMA)) {
+                // URI
+                URI value = URI.create((String) entry.getValue());
+                writer.withContextAttribute(key, value);
+            } else {
+                // String
+                writer.withContextAttribute(key, (String) entry.getValue());
+            }
+        }
 
+        // TOOD: data
+        return writer.end();
     }
 
 }

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

@@ -22,6 +22,7 @@
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.CloudEventData;
+import io.cloudevents.core.v1.CloudEventV1;
 import io.cloudevents.AvroCloudEvent;
 import io.cloudevents.AvroCloudEventData;
 
@@ -32,14 +33,17 @@ public static final AvroCloudEvent toAvro(CloudEvent e) {
 
         Map<CharSequence, Object> attrs = new HashMap<>();
 
-        attrs.put("type", e.getType());
-        attrs.put("specversion", e.getSpecVersion().toString());
-        attrs.put("id", e.getId());
-        attrs.put("source", e.getSource());
-        attrs.put("time", e.getTime());
-        attrs.put("dataschema", e.getDataSchema());
-        attrs.put("contenttype", AvroFormat.AVRO_CONTENT_TYPE);
-        attrs.put("datacontenttype", e.getDataContentType());
+        attrs.put(CloudEventV1.TYPE, e.getType());
+        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
+        attrs.put(CloudEventV1.DATASCHEMA, e.getDataSchema().toString());
+        attrs.put(CloudEventV1.SUBJECT, e.getSubject());
+
+        attrs.put(CloudEventV1.DATACONTENTTYPE, e.getDataContentType());
 
         avroCloudEvent.setAttribute(attrs);