Merge pull request #79 from ruromero/base64

Fix base64 marshalling

CHANGELOG.md

@@ -6,6 +6,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ## [Unreleased]
 
+- Improved base64 marshalling: PR [#79](https://github.com/cloudevents/sdk-java/pull/79)
+
 ## [1.1.0]
 
 ### Added

README.md

@@ -77,12 +77,12 @@ final String eventType = "My.Cloud.Event.Type";
 final byte[] payload = "a-binary-event-data".getBytes();
 
 // passing in the given attributes
-final CloudEventImpl<byte[]> cloudEvent =
-  CloudEventBuilder.<byte[]>builder()
+final CloudEventImpl<String> cloudEvent =
+  CloudEventBuilder.<String>builder()
     .withType(eventType)
     .withId(eventId)
     .withSource(src)
-    .withData(payload)
+    .withDataBase64(payload)
     .build();
 
 // marshalling as json that will have the data_base64

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

@@ -1,12 +1,12 @@
 /**
  * Copyright 2019 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.
@@ -20,27 +20,26 @@ import java.util.Optional;
 
 /**
  * An abstract event envelope
- * 
- * @author fabiojose
- *
  * @param <A> The attributes type
  * @param <T> The 'data' type
+ * @author fabiojose
  */
 public interface CloudEvent<A extends Attributes, T> {
 
-	/**
-	 * The event context attributes
-	 */
-	A getAttributes();
-	
-	/**
-	 * The event data
-	 */
-	Optional<T> getData();
-	
-	/**
-	 * The event extensions
-	 */
-	Map<String, Object> getExtensions();
-	
+    /**
+     * The event context attributes
+     */
+    A getAttributes();
+
+    /**
+     * The event data
+     */
+    Optional<T> getData();
+
+    byte[] getDataBase64();
+
+    /**
+     * The event extensions
+     */
+    Map<String, Object> getExtensions();
 }

api/src/main/java/io/cloudevents/json/Json.java

@@ -55,7 +55,6 @@ public final class Json {
         try {
             return MAPPER.writeValueAsString(obj);
         } catch (Exception e) {
-        	e.printStackTrace();
             throw new IllegalStateException("Failed to encode as JSON: " + e.getMessage());
         }
     }
@@ -71,7 +70,6 @@ public final class Json {
         try {
             return MAPPER.writeValueAsBytes(obj);
         } catch (Exception e) {
-        	e.printStackTrace();
             throw new IllegalStateException("Failed to encode as JSON: " + e.getMessage());
         }
     }
@@ -224,31 +222,18 @@ public final class Json {
      */
     public static <T, A extends Attributes> DataUnmarshaller<String, T, A> 
     umarshaller(Class<T> type) {
-    	return new DataUnmarshaller<String, T, A>() {
-			@Override
-			public T unmarshal(String payload, A attributes) {
-				return Json.decodeValue(payload, type);
-			}
-		};
+    	return (payload, attributes) -> Json.decodeValue(payload, type);
     }
     
     /**
      * Unmarshals a byte array into T type
      * @param <T> The 'data' type
      * @param <A> The attributes type
-     * @param payload The byte array
-     * @param attribues
      * @return The data objects
      */
     public static <T, A extends Attributes> DataUnmarshaller<byte[], T, A> 
     		binaryUmarshaller(Class<T> type) {
-    	
-    	return new DataUnmarshaller<byte[], T, A>() {
-			@Override
-			public T unmarshal(byte[] payload, A attributes) {
-				return Json.binaryDecodeValue(payload, type);
-			}
-		};
+    	return (payload, attributes) -> Json.binaryDecodeValue(payload, type);
     }
     
     /**
@@ -258,12 +243,7 @@ public final class Json {
      * @return A new instance of {@link DataMarshaller}
      */
     public static <T, H> DataMarshaller<String, T, H> marshaller() {
-    	return new DataMarshaller<String, T, H>() {
-			@Override
-			public String marshal(T data, Map<String, H> headers) {
-				return Json.encode(data);
-			}
-		};
+    	return (data, headers) -> Json.encode(data);
     }
     
     /**

api/src/main/java/io/cloudevents/v02/CloudEventImpl.java

@@ -40,9 +40,9 @@ import io.cloudevents.extensions.InMemoryFormat;
 
 /**
  * The event implementation
- * 
+ *
  * @author fabiojose
- * 
+ *
  */
 @JsonInclude(value = Include.NON_ABSENT)
 public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
@@ -50,61 +50,68 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 	@JsonIgnore
 	@NotNull
 	private final AttributesImpl attributes;
-	
+
 	private final T data;
-	
+
 	@NotNull
 	private final Map<String, Object> extensions;
-	
+
 	private final Set<ExtensionFormat> extensionsFormats;
 
 	CloudEventImpl(AttributesImpl attributes, T data,
 			Set<ExtensionFormat> extensions) {
-		
+
 		this.attributes = attributes;
-		
+
 		this.data = data;
-		
+
 		this.extensions = extensions.stream()
 			.map(ExtensionFormat::memory)
 			.collect(Collectors.toMap(InMemoryFormat::getKey,
 					InMemoryFormat::getValue));
-		
+
 		this.extensionsFormats = extensions;
 	}
-	
+
 	/**
 	 * Used by the {@link Accessor} to access the set of {@link ExtensionFormat}
 	 */
 	Set<ExtensionFormat> getExtensionsFormats() {
 		return extensionsFormats;
 	}
-	
-	@JsonUnwrapped
-	@Override
-	public AttributesImpl getAttributes() {
-		return this.attributes;
-	}
-	
+
+    @Override
+    @JsonUnwrapped
+    public AttributesImpl getAttributes() {
+        return this.attributes;
+    }
+
 	/**
      * The event payload. The payload depends on the eventType,
      * schemaURL and eventTypeVersion, the payload is encoded into
      * a media format which is specified by the contentType attribute
      * (e.g. application/json).
      */
+	@Override
 	public Optional<T> getData() {
 		return Optional.ofNullable(data);
 	}
 
+	@Override
+	public byte[] getDataBase64() {
+		return null;
+	}
+
+	@Override
 	@JsonAnyGetter
 	public Map<String, Object> getExtensions() {
 		return Collections.unmodifiableMap(extensions);
 	}
-	
+
 	/**
 	 * The unique method that allows mutation. Used by
 	 * Jackson Framework to inject the extensions.
-	 * 
+	 *
 	 * @param name Extension name
 	 * @param value Extension value
 	 */
@@ -112,7 +119,7 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 	void addExtension(String name, Object value) {
 		extensions.put(name, value);
 	}
-	
+
 	@JsonCreator
 	public static <T> CloudEventImpl<T> build(
 			@JsonProperty("id") String id,
@@ -122,7 +129,7 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 			@JsonProperty("schemaurl") URI schemaurl,
 			@JsonProperty("contenttype") String contenttype,
 			@JsonProperty("data") T data) {
-		
+
 		return CloudEventBuilder.<T>builder()
 				.withId(id)
 				.withSource(source)

api/src/main/java/io/cloudevents/v02/http/Unmarshallers.java

@@ -1,6 +1,5 @@
 package io.cloudevents.v02.http;
 
-import javax.validation.Valid;
 import javax.validation.Validator;
 
 import io.cloudevents.extensions.DistributedTracingExtension;
@@ -93,15 +92,10 @@ public class Unmarshallers {
 			.next()
 			.map((payload, extensions) -> {			
 				CloudEventImpl<T> event =
-					Json.<CloudEventImpl<T>>
-						decodeValue(payload, CloudEventImpl.class, typeOfData);
-				
+					Json.decodeValue(payload, CloudEventImpl.class, typeOfData);
 				CloudEventBuilder<T> builder = 
-					CloudEventBuilder.<T>builder(event);
-				
-				extensions.get().forEach(extension -> {
-					builder.withExtension(extension);
-				});
+					CloudEventBuilder.builder(event);
+				extensions.get().forEach(builder::withExtension);
 
 				return builder.withValidator(validator).build();
 			});

api/src/main/java/io/cloudevents/v03/CloudEventImpl.java

@@ -46,7 +46,7 @@ import io.cloudevents.extensions.InMemoryFormat;
  */
 @JsonInclude(value = Include.NON_ABSENT)
 public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
-	
+
 	@JsonIgnore
 	@NotNull
 	private final AttributesImpl attributes;
@@ -78,8 +78,8 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 		return extensionsFormats;
 	}
 
-	@JsonUnwrapped
 	@Override
+	@JsonUnwrapped
 	public AttributesImpl getAttributes() {
 		return attributes;
 	}
@@ -89,8 +89,13 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 		return Optional.ofNullable(data);
 	}
 
-	@JsonAnyGetter
 	@Override
+	public byte[] getDataBase64() {
+		return null;
+	}
+
+	@Override
+	@JsonAnyGetter
 	public Map<String, Object> getExtensions() {
 		return Collections.unmodifiableMap(extensions);
 	}

api/src/main/java/io/cloudevents/v1/CloudEventBuilder.java

@@ -1,13 +1,10 @@
 package io.cloudevents.v1;
 
-import static java.lang.String.format;
-
 import java.net.URI;
 import java.time.ZonedDateTime;
 import java.util.Collection;
 import java.util.HashSet;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -19,6 +16,8 @@ import io.cloudevents.CloudEvent;
 import io.cloudevents.extensions.ExtensionFormat;
 import io.cloudevents.fun.EventBuilder;
 
+import static java.lang.String.format;
+
 /**
  * 
  * @author fabiojose
@@ -46,6 +45,7 @@ public class CloudEventBuilder<T> implements
 	private ZonedDateTime time;
 
 	private T data;
+	private byte[] dataBase64;
 
 	private final Set<ExtensionFormat> extensions = new HashSet<>();
 
@@ -90,7 +90,9 @@ public class CloudEventBuilder<T> implements
 		attributes.getTime().ifPresent(result::withTime);
 		Accessor.extensionsOf(base).forEach(result::withExtension);
 		base.getData().ifPresent(result::withData);
-
+		if(base.getDataBase64() != null) {
+			result.withDataBase64(base.getDataBase64());
+		}
 		return result;
 	}
 
@@ -112,6 +114,7 @@ public class CloudEventBuilder<T> implements
 
 		return builder
 			.withData(data)
+			.withDataBase64(dataBase64)
 			.withValidator(validator)
 			.build();
 	}
@@ -127,31 +130,28 @@ public class CloudEventBuilder<T> implements
 		AttributesImpl attributes = new AttributesImpl(id, source, SPEC_VERSION, type,
 				datacontenttype, dataschema, subject, time);
 		
-		CloudEventImpl<T> cloudEvent =
-			new CloudEventImpl<>(attributes, data, extensions);
-		
-		if(data instanceof byte[]) {
-			cloudEvent.setDataBase64((byte[])data);
+		CloudEventImpl<T> cloudEvent;
+		if(data != null) {
+			cloudEvent = new CloudEventImpl<>(attributes, data, extensions);
+		} else {
+			cloudEvent = new CloudEventImpl<>(attributes, dataBase64, extensions);
 		}
+
 		if(validator == null) {
 			validator = getValidator();
 		}
-		Set<ConstraintViolation<Object>> violations =
-				validator.validate(cloudEvent);
-		
+		Set<ConstraintViolation<Object>> violations = new HashSet<>();
+		violations.addAll(validator.validate(cloudEvent));
 		violations.addAll(validator.validate(cloudEvent.getAttributes()));
 		
 		final String errs = 
 			violations.stream()
 				.map(v -> format(MESSAGE, v.getPropertyPath(), v.getMessage()))
 				.collect(Collectors.joining(MESSAGE_SEPARATOR));
-		
-		Optional.ofNullable(
-			"".equals(errs) ? null : errs
-					
-		).ifPresent((e) -> {
-			throw new IllegalStateException(format(ERR_MESSAGE, e));
-		});
+
+		if(!errs.trim().isEmpty()) {
+			throw new IllegalStateException(format(ERR_MESSAGE, errs));
+		}
 		
 		return cloudEvent;
 	}
@@ -198,7 +198,12 @@ public class CloudEventBuilder<T> implements
 		this.data = data;
 		return this;
 	}
-	
+
+	public CloudEventBuilder<T> withDataBase64(byte[] dataBase64) {
+		this.dataBase64 = dataBase64;
+		return this;
+	}
+
 	public CloudEventBuilder<T> withExtension(ExtensionFormat extension) {
 		this.extensions.add(extension);
 		return this;

api/src/main/java/io/cloudevents/v1/CloudEventImpl.java

@@ -18,7 +18,6 @@ package io.cloudevents.v1;
 import java.net.URI;
 import java.time.ZonedDateTime;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -26,106 +25,95 @@ import java.util.stream.Collectors;
 
 import javax.validation.constraints.NotNull;
 
-import com.fasterxml.jackson.annotation.JsonAlias;
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonUnwrapped;
-
 import io.cloudevents.CloudEvent;
 import io.cloudevents.extensions.ExtensionFormat;
 import io.cloudevents.extensions.InMemoryFormat;
 
 /**
- * 
+ *
  * @author fabiojose
  * @version 1.0
  */
-@JsonInclude(value = Include.NON_ABSENT)
+@JsonInclude(value = JsonInclude.Include.NON_ABSENT)
 public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
-	
-	public static final String EVENT_DATA_FIELD = "data";
-	public static final String EVENT_DATA_BASE64_FILED = "data_base64";
-	
-	@JsonIgnore
+
 	@NotNull
+	@JsonIgnore
 	private final AttributesImpl attributes;
-	
-	@JsonIgnore
+
 	private final T data;
-	
+
 	//To use with json binary data
-	@JsonIgnore
-	private byte[] dataBase64;
-	
+	private final byte[] dataBase64;
+
 	@NotNull
 	private final Map<String, Object> extensions;
-	
+
 	private final Set<ExtensionFormat> extensionsFormats;
-	
+
+	CloudEventImpl(AttributesImpl attributes, byte[] dataBase64,
+				   Set<ExtensionFormat> extensions){
+		this(attributes, extensions, null, dataBase64);
+	}
+
 	CloudEventImpl(AttributesImpl attributes, T data,
 			Set<ExtensionFormat> extensions){
+		this(attributes, extensions, data, null);
+	}
+
+	private CloudEventImpl(AttributesImpl attributes, Set<ExtensionFormat> extensions, T data, byte[] dataBase64){
 		this.attributes = attributes;
-		this.data = data;
-		
 		this.extensions = extensions.stream()
-				.map(ExtensionFormat::memory)
-				.collect(Collectors.toMap(InMemoryFormat::getKey,
-						InMemoryFormat::getValue));
-		
+			.map(ExtensionFormat::memory)
+			.collect(Collectors.toMap(InMemoryFormat::getKey,
+									  InMemoryFormat::getValue));
+		this.data = data;
+		this.dataBase64 = dataBase64;
 		this.extensionsFormats = extensions;
 	}
-	
+
 	/**
 	 * Used by the {@link Accessor} to access the set of {@link ExtensionFormat}
 	 */
 	Set<ExtensionFormat> getExtensionsFormats() {
 		return extensionsFormats;
 	}
-	
-	/**
-	 * To handle the JSON base64 serialization
-	 * @param data The byte array to encode as base64
-	 */
-	void setDataBase64(byte[] data) {
-		this.dataBase64 = data;
-	}
-	
-	@JsonUnwrapped
+
+
 	@Override
+	@JsonUnwrapped
 	public AttributesImpl getAttributes() {
 		return attributes;
 	}
 
 	@Override
-	@JsonIgnore
 	public Optional<T> getData() {
 		return Optional.ofNullable(data);
 	}
 
-	@JsonAnyGetter
 	@Override
-	public Map<String, Object> getExtensions() {
-		Map<String, Object> result = new HashMap<>(extensions);
-		
-		if(null== dataBase64) {
-			if(null!= data) {
-				result.put(EVENT_DATA_FIELD, data);
-			}
-		} else {
-			result.put(EVENT_DATA_BASE64_FILED, dataBase64);
-		}
-		return Collections.unmodifiableMap(result);
+	@JsonProperty("data_base64")
+	public byte[] getDataBase64() {
+		return dataBase64;
 	}
-	
+
+	@Override
+	@JsonAnyGetter
+	public Map<String, Object> getExtensions() {
+		return Collections.unmodifiableMap(extensions);
+	}
+
 	/**
 	 * The unique method that allows mutation. Used by
 	 * Jackson Framework to inject the extensions.
-	 * 
+	 *
 	 * @param name Extension name
 	 * @param value Extension value
 	 */
@@ -146,10 +134,9 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 			@JsonProperty("dataschema") URI dataschema,
 			@JsonProperty("subject") String subject,
 			@JsonProperty("time") ZonedDateTime time,
-			@JsonProperty("data")
-			@JsonAlias("data_base64")
-			T data){
-		
+			@JsonProperty("data") T data,
+			@JsonProperty("data_base64") byte[] dataBase64){
+
 		return CloudEventBuilder.<T>builder()
 				.withId(id)
 				.withSource(source)
@@ -158,6 +145,7 @@ public class CloudEventImpl<T> implements CloudEvent<AttributesImpl, T> {
 				.withDataschema(dataschema)
 				.withDataContentType(datacontenttype)
 				.withData(data)
+				.withDataBase64(dataBase64)
 				.withSubject(subject)
 				.build();
 	}

api/src/main/java/io/cloudevents/v1/http/Marshallers.java

@@ -23,7 +23,7 @@ import io.cloudevents.v1.http.HeaderMapper;
 public class Marshallers {
 	private Marshallers() {}
 	
-	private static final Map<String, String> NO_HEADERS = 
+	private static final Map<String, String> NO_HEADERS =
 			new HashMap<String, String>();
 		
 	/**
@@ -43,7 +43,7 @@ public class Marshallers {
 				.map(ExtensionFormat::marshal)
 				.map(HeaderMapper::map)
 				.map(Json.<T, String>marshaller()::marshal)
-				.builder(Wire<String, String, String>::new);
+				.builder(Wire::new);
 	}
 	
 	/**
@@ -58,10 +58,8 @@ public class Marshallers {
 		StructuredMarshaller.
 		  <AttributesImpl, T, String, String>builder()
 			.mime("Content-Type", "application/cloudevents+json")
-			.map((event) -> {
-				return Json.<CloudEvent<AttributesImpl, T>, String>
-							marshaller().marshal(event, NO_HEADERS);
-			})
+			.map((event) -> Json.<CloudEvent<AttributesImpl, T>, String>
+						marshaller().marshal(event, NO_HEADERS))
 			.map(Accessor::extensionsOf)
 			.map(ExtensionFormat::marshal)
 			.map(HeaderMapper::map);

api/src/test/java/io/cloudevents/v1/CloudEventJacksonTest.java

@@ -15,12 +15,6 @@
  */
 package io.cloudevents.v1;
 
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
 import java.io.InputStream;
 import java.net.URI;
 import java.time.ZonedDateTime;
@@ -28,20 +22,21 @@ import java.util.Base64;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.ExpectedException;
-
 import com.fasterxml.jackson.core.type.TypeReference;
-
 import io.cloudevents.CloudEvent;
 import io.cloudevents.extensions.DistributedTracingExtension;
 import io.cloudevents.extensions.ExtensionFormat;
 import io.cloudevents.json.Json;
 import io.cloudevents.json.types.Much;
-import io.cloudevents.v1.AttributesImpl;
-import io.cloudevents.v1.CloudEventBuilder;
-import io.cloudevents.v1.CloudEventImpl;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 /**
  * 
@@ -69,8 +64,7 @@ public class CloudEventJacksonTest {
 		
 		// act
 		String json = Json.encode(ce);
-		System.out.println(json);
-		
+
 		// assert
 		assertTrue(json.contains("x10"));
 		assertTrue(json.contains("/source"));
@@ -110,7 +104,6 @@ public class CloudEventJacksonTest {
 		assertTrue(json.contains("datacontenttype"));
 		assertTrue(json.contains("\"subject\""));
 		
-		System.out.println(json);
 		Pattern pat = Pattern.compile("(\"data\")");
 		Matcher mat = pat.matcher(json);
 		int counter = 0;
@@ -301,15 +294,13 @@ public class CloudEventJacksonTest {
 		byte[] expected = "mydata".getBytes();
 		
 		// act
-		CloudEvent<AttributesImpl, byte[]> ce = 
+		CloudEvent<AttributesImpl, String> ce =
 	        	Json.fromInputStream(resourceOf("1_base64.json"), 
-	        			new TypeReference<CloudEventImpl<byte[]>>() {});
-		
-		System.out.println(new String(ce.getData().get()));
-        
+	        			new TypeReference<CloudEventImpl<String>>() {});
+
         // assert
-		assertTrue(ce.getData().isPresent());
-		assertArrayEquals(expected, ce.getData().get());
+		assertNotNull(ce.getDataBase64());
+		assertArrayEquals(expected, ce.getDataBase64());
     }
 	
 	@Test
@@ -326,8 +317,8 @@ public class CloudEventJacksonTest {
 		String expected = 
 			Base64.getEncoder().encodeToString(data);
 		
-		CloudEventImpl<byte[]> event = 
-			CloudEventBuilder.<byte[]>builder()
+		CloudEventImpl<String> event =
+			CloudEventBuilder.<String>builder()
 				.withId("0xbin")
 				.withSource(URI.create("/customers/445"))
 				.withType("customers.ordering")
@@ -335,7 +326,7 @@ public class CloudEventJacksonTest {
 				.withDataschema(URI.create("http://schame.server.com/customer/order"))
 				.withSubject("orders.json")
 				.withTime(ZonedDateTime.now())
-				.withData(data)
+				.withDataBase64(data)
 				.build();
 
 		// act

api/src/test/java/io/cloudevents/v1/http/HTTPBinaryUnmarshallerTest.java

@@ -15,21 +15,19 @@
  */
 package io.cloudevents.v1.http;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
 
-import org.junit.Test;
-
 import io.cloudevents.CloudEvent;
 import io.cloudevents.extensions.DistributedTracingExtension;
 import io.cloudevents.json.types.Much;
 import io.cloudevents.v1.AttributesImpl;
-import io.cloudevents.v1.http.Unmarshallers;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 /**
  * 
@@ -56,7 +54,7 @@ public class HTTPBinaryUnmarshallerTest {
 		String payload = "{\"wow\":\"yes!\"}";
 		
 		// act
-		CloudEvent<AttributesImpl, Much> actual = 
+		CloudEvent<AttributesImpl, Much> actual =
 			Unmarshallers.binary(Much.class)
 				.withHeaders(() -> myHeaders)
 				.withPayload(() -> payload)
@@ -99,7 +97,7 @@ public class HTTPBinaryUnmarshallerTest {
 		String payload = "{\"wow\":\"yes!\"}";
 		
 		// act
-		CloudEvent<AttributesImpl, Much> actual = 
+		CloudEvent<AttributesImpl, Much> actual =
 			Unmarshallers.binary(Much.class)
 				.withHeaders(() -> myHeaders)
 				.withPayload(() -> payload)

api/src/test/java/io/cloudevents/v1/http/HTTPStructuredUnmarshallerTest.java

@@ -16,9 +16,12 @@
 package io.cloudevents.v1.http;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertFalse;
 
 import java.net.URI;
+import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -128,7 +131,51 @@ public class HTTPStructuredUnmarshallerTest {
 		assertTrue(actual.getData().isPresent());
 		assertEquals(expected.getData().get(), actual.getData().get());
 	}
-	
+
+	@Test
+	public void should_unmarshal_json_envelope_and_text_data_base64() {
+		// setup
+		Map<String, Object> httpHeaders = new HashMap<>();
+		httpHeaders.put("Content-Type", "application/cloudevents+json");
+
+		String ceData = "yes!";
+		byte[] base64Data = Base64.getEncoder().encode(ceData.getBytes());
+		String json = "{\"data_base64\":\"" + new String(base64Data) + "\",\"id\":\"x10\",\"source\":\"/source\",\"specversion\":\"1.0\",\"type\":\"event-type\",\"datacontenttype\":\"text/plain\"}";
+
+		CloudEventImpl<String> expected =
+			CloudEventBuilder.<String>builder()
+				.withId("x10")
+				.withSource(URI.create("/source"))
+				.withType("event-type")
+				.withDataContentType("text/plain")
+				.withDataBase64(ceData.getBytes())
+				.build();
+
+		// act
+		CloudEvent<AttributesImpl, String> actual =
+			Unmarshallers.structured(String.class)
+				.withHeaders(() -> httpHeaders)
+				.withPayload(() -> json)
+				.unmarshal();
+
+		// assert
+		assertEquals(expected.getAttributes().getSpecversion(),
+					 actual.getAttributes().getSpecversion());
+
+		assertEquals(expected.getAttributes().getId(),
+					 actual.getAttributes().getId());
+
+		assertEquals(expected.getAttributes().getSource(),
+					 actual.getAttributes().getSource());
+
+		assertEquals(expected.getAttributes().getType(),
+					 actual.getAttributes().getType());
+
+		assertFalse(actual.getData().isPresent());
+		assertNotNull(actual.getDataBase64());
+		assertEquals(new String(expected.getDataBase64()), new String(actual.getDataBase64()));
+	}
+
 	@Test
 	public void should_unmarshal_the_tracing_extension_from_headers() {
 		// setup

kafka/src/main/java/io/cloudevents/v02/kafka/Marshallers.java

@@ -58,7 +58,7 @@ public class Marshallers {
 				.map(ExtensionFormat::marshal)
 				.map(HeaderMapper::map)
 				.map(Json::binaryMarshal)
-				.builder(Wire<byte[], String, byte[]>::new);
+				.builder(Wire::new);
 	}
 
 	/**