Add support for Message<CloudEvent> (#315)

* Add support for Message<CloudEvent>

Signed-off-by: Dave Syer <[email protected]>

* Add support for structured messages with Spring Message<?>

Signed-off-by: Dave Syer <[email protected]>

* Push private classes out to shared utilities

Signed-off-by: Dave Syer <[email protected]>

* Resolve some more review comments

Signed-off-by: Dave Syer <[email protected]>

* Restructure MessageReader and MessageWriter

Signed-off-by: Dave Syer <[email protected]>

* Remove integration test (depends on snapshots still)

Signed-off-by: Dave Syer <[email protected]>

* Simplify message converter but drop support for structured format

Signed-off-by: Dave Syer <[email protected]>

* Make HTTP optional

Signed-off-by: Dave Syer <[email protected]>

* Drop snapshot reporitory declarations

Signed-off-by: Dave Syer <[email protected]>

Author: Dave Syer

spring/pom.xml

@@ -58,6 +58,11 @@
             <artifactId>spring-webflux</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-messaging</artifactId>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>io.cloudevents</groupId>
             <artifactId>cloudevents-core</artifactId>
@@ -67,7 +72,7 @@
             <groupId>io.cloudevents</groupId>
             <artifactId>cloudevents-http-basic</artifactId>
             <version>${project.version}</version>
-            <!-- This will be optional too if we want to support messaging -->
+            <optional>true</optional>
         </dependency>
         <dependency>
             <groupId>org.apache.tomcat.embed</groupId>
@@ -101,5 +106,6 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
+
 </project>
 

spring/src/main/java/io/cloudevents/spring/messaging/CloudEventHeaderUtils.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.spring.messaging;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.CloudEventContext;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.CloudEventUtils;
+
+import org.springframework.messaging.MessageHeaders;
+
+import static io.cloudevents.spring.messaging.CloudEventsHeaders.CE_PREFIX;
+
+/**
+ * Utility class for copying message headers to and from {@link CloudEventContext}.
+ * 
+ * @author Dave Syer
+ *
+ */
+public class CloudEventHeaderUtils {
+
+	/**
+	 * Helper method for converting {@link MessageHeaders} to a {@link CloudEventContext}.
+	 * The input headers must represent a valid event in "binary" form, i.e. it must have
+	 * headers "ce-id", "ce-specversion" etc.
+	 * @param headers the input request headers
+	 * @return a {@link CloudEventContext} that can be used to create a new
+	 * {@link CloudEvent}
+	 * 
+	 */
+	public static CloudEventContext fromMap(Map<String, Object> headers) {
+		Object value = headers.get(CloudEventsHeaders.SPEC_VERSION);
+		SpecVersion version = value == null ? SpecVersion.V1 : SpecVersion.parse(value.toString());
+		return CloudEventUtils.toEvent(new MessageBinaryMessageReader(version, headers));
+	}
+
+	/**
+	 * Helper method for extracting {@link MessageHeaders} from a {@link CloudEvent}. The
+	 * result will contain headers canonicalized with a "ce-" prefix, analogous to the
+	 * "binary" message format in Cloud Events.
+	 * @param event the input {@link CloudEvent}
+	 * @return the response headers represented by the event
+	 */
+	public static Map<String, Object> toMap(CloudEvent event) {
+		Map<String, Object> headers = new HashMap<>();
+		// Probably this should be done in CloudEventContextReaderAdapter
+		headers.put(CE_PREFIX + "specversion", event.getSpecVersion().toString());
+		MessageBuilderMessageWriter writer = new MessageBuilderMessageWriter(headers);
+		CloudEventUtils.toContextReader(event).readContext(writer);
+		return writer.end().getHeaders();
+	}
+
+}

spring/src/main/java/io/cloudevents/spring/messaging/CloudEventMessageConverter.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.spring.messaging;
+
+import java.nio.charset.Charset;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.CloudEventContext;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.CloudEventUtils;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.core.message.impl.MessageUtils;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
+
+/**
+ * A {@link MessageConverter} that can translate to and from a {@link Message
+ * Message<byte[]>} or {@link Message Message<String>} and a {@link CloudEvent}. The
+ * {@link CloudEventContext} is canonicalized, with key names given a "ce-" prefix in the
+ * {@link MessageHeaders}.
+ * 
+ * @author Dave Syer
+ *
+ */
+public class CloudEventMessageConverter implements MessageConverter {
+
+	@Override
+	public Object fromMessage(Message<?> message, Class<?> targetClass) {
+		if (CloudEvent.class.isAssignableFrom(targetClass)) {
+			return createMessageReader(message).toEvent();
+		}
+		return null;
+	}
+
+	@Override
+	public Message<?> toMessage(Object payload, MessageHeaders headers) {
+		if (payload instanceof CloudEvent) {
+			CloudEvent event = (CloudEvent) payload;
+			return CloudEventUtils.toReader(event).read(new MessageBuilderMessageWriter(headers));
+		}
+		return null;
+	}
+
+	private MessageReader createMessageReader(Message<?> message) {
+		return MessageUtils.parseStructuredOrBinaryMessage( //
+				() -> contentType(message.getHeaders()), //
+				format -> structuredMessageReader(message, format), //
+				() -> version(message.getHeaders()), //
+				version -> binaryMessageReader(message, version) //
+		);
+	}
+
+	private String version(MessageHeaders message) {
+		if (message.containsKey(CloudEventsHeaders.SPEC_VERSION)) {
+			return message.get(CloudEventsHeaders.SPEC_VERSION).toString();
+		}
+		return null;
+	}
+
+	private MessageReader binaryMessageReader(Message<?> message, SpecVersion version) {
+		return new MessageBinaryMessageReader(version, message.getHeaders(), getBinaryData(message));
+	}
+
+	private MessageReader structuredMessageReader(Message<?> message, EventFormat format) {
+		return new GenericStructuredMessageReader(format, getBinaryData(message));
+	}
+
+	private String contentType(MessageHeaders message) {
+		if (message.containsKey(MessageHeaders.CONTENT_TYPE)) {
+			return message.get(MessageHeaders.CONTENT_TYPE).toString();
+		}
+		if (message.containsKey(CloudEventsHeaders.CONTENT_TYPE)) {
+			return message.get(CloudEventsHeaders.CONTENT_TYPE).toString();
+		}
+		return null;
+	}
+
+	private byte[] getBinaryData(Message<?> message) {
+		Object payload = message.getPayload();
+		if (payload instanceof byte[]) {
+			return (byte[]) payload;
+		}
+		else if (payload instanceof String) {
+			return ((String) payload).getBytes(Charset.defaultCharset());
+		}
+		return null;
+	}
+
+}

spring/src/main/java/io/cloudevents/spring/messaging/CloudEventsHeaders.java

@@ -0,0 +1,28 @@
+/*
+ * 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.spring.messaging;
+
+public class CloudEventsHeaders {
+
+	public static final String CE_PREFIX = "ce-";
+
+	public static final String SPEC_VERSION = CE_PREFIX + "specversion";
+
+	public static final String CONTENT_TYPE = CE_PREFIX + "datacontenttype";
+
+}

spring/src/main/java/io/cloudevents/spring/messaging/MessageBinaryMessageReader.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.spring.messaging;
+
+import java.util.Map;
+import java.util.function.BiConsumer;
+
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.data.BytesCloudEventData;
+import io.cloudevents.core.message.impl.BaseGenericBinaryMessageReaderImpl;
+
+import static io.cloudevents.spring.messaging.CloudEventsHeaders.CE_PREFIX;
+import static org.springframework.messaging.MessageHeaders.CONTENT_TYPE;
+
+/**
+ * Utility for converting maps (message headers) to `CloudEvent` contexts.
+ * 
+ * @author Dave Syer
+ *
+ */
+class MessageBinaryMessageReader extends BaseGenericBinaryMessageReaderImpl<String, Object> {
+
+	private final Map<String, Object> headers;
+
+	public MessageBinaryMessageReader(SpecVersion version, Map<String, Object> headers, byte[] payload) {
+		super(version, payload == null ? null : BytesCloudEventData.wrap(payload));
+		this.headers = headers;
+	}
+
+	public MessageBinaryMessageReader(SpecVersion version, Map<String, Object> headers) {
+		this(version, headers, null);
+	}
+
+	@Override
+	protected boolean isContentTypeHeader(String key) {
+		return CONTENT_TYPE.equalsIgnoreCase(key);
+	}
+
+	@Override
+	protected boolean isCloudEventsHeader(String key) {
+		return key != null && key.length() > 3
+				&& key.substring(0, CE_PREFIX.length()).toLowerCase().startsWith(CE_PREFIX);
+	}
+
+	@Override
+	protected String toCloudEventsKey(String key) {
+		return key.substring(CE_PREFIX.length()).toLowerCase();
+	}
+
+	@Override
+	protected void forEachHeader(BiConsumer<String, Object> fn) {
+		headers.forEach((k, v) -> fn.accept(k, v));
+	}
+
+	@Override
+	protected String toCloudEventsValue(Object value) {
+		return value.toString();
+	}
+
+}

spring/src/main/java/io/cloudevents/spring/messaging/MessageBuilderMessageWriter.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2019-2019 the original author or authors.
+ *
+ * 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
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.spring.messaging;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import io.cloudevents.CloudEventData;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.rw.CloudEventContextWriter;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * Internal utility class for copying <code>CloudEvent</code> context to a map (message
+ * headers).
+ * 
+ * @author Dave Syer
+ *
+ */
+class MessageBuilderMessageWriter
+		implements CloudEventWriter<Message<byte[]>>, MessageWriter<MessageBuilderMessageWriter, Message<byte[]>> {
+
+	private Map<String, Object> headers = new HashMap<>();
+
+	public MessageBuilderMessageWriter(Map<String, Object> headers) {
+		this.headers.putAll(headers);
+	}
+
+	public MessageBuilderMessageWriter() {
+	}
+
+	@Override
+	public Message<byte[]> setEvent(EventFormat format, byte[] value) throws CloudEventRWException {
+		headers.put(CloudEventsHeaders.CONTENT_TYPE, format.serializedContentType());
+		return MessageBuilder.withPayload(value).copyHeaders(headers).build();
+	}
+
+	@Override
+	public Message<byte[]> end(CloudEventData value) throws CloudEventRWException {
+		return MessageBuilder.withPayload(value == null ? new byte[0] : value.toBytes()).copyHeaders(headers).build();
+	}
+
+	@Override
+	public Message<byte[]> end() {
+		return MessageBuilder.withPayload(new byte[0]).copyHeaders(headers).build();
+	}
+
+	@Override
+	public CloudEventContextWriter withContextAttribute(String name, String value) throws CloudEventRWException {
+		headers.put(CloudEventsHeaders.CE_PREFIX + name, value);
+		return this;
+	}
+
+	@Override
+	public MessageBuilderMessageWriter create(SpecVersion version) {
+		headers.put(CloudEventsHeaders.SPEC_VERSION, version.toString());
+		return this;
+	}
+
+}

spring/src/main/java/io/cloudevents/spring/messaging/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes related to working with Cloud Events within the context of Spring Messaging.
+ */
+package io.cloudevents.spring.messaging;