+ * 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 + *
+ * http://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.mqtt.core; + +import io.cloudevents.SpecVersion; +import io.cloudevents.core.data.BytesCloudEventData; +import io.cloudevents.core.message.impl.BaseGenericBinaryMessageReaderImpl; +import io.cloudevents.core.v1.CloudEventV1; + +import java.util.function.BiConsumer; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Enable the hydration of a CloudEvent in binary mode from an MQTT message. + *
+ * This abstract class provides common behavior across different MQTT
+ * client implementations.
+ */
+public abstract class BaseMqttBinaryMessageReader extends BaseGenericBinaryMessageReaderImpl
+ * 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
+ *
+ * http://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.mqtt.core;
+
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.provider.EventFormatProvider;
+
+/**
+ * General MQTT Utilities and Helpers
+ */
+public class MqttUtils {
+
+ private static final String DEFAULT_FORMAT = "application/cloudevents+json";
+
+ private MqttUtils() {
+ }
+
+ /**
+ * Obtain the {@link EventFormat} to use when working with MQTT V3
+ * messages.
+ *
+ * @return An event format.
+ */
+ public static EventFormat getDefaultEventFormat() {
+
+ return EventFormatProvider.getInstance().resolveFormat(DEFAULT_FORMAT);
+
+ }
+
+ /**
+ * Get the default content type to assume for MQTT messages.
+ *
+ * @return A Content-Type
+ */
+ public static final String getDefaultContentType() {
+ return DEFAULT_FORMAT;
+ }
+
+}
diff --git a/bindings/mqtt/hivemq/pom.xml b/bindings/mqtt/hivemq/pom.xml
new file mode 100644
index 000000000..e3106fca3
--- /dev/null
+++ b/bindings/mqtt/hivemq/pom.xml
@@ -0,0 +1,80 @@
+
+
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.mqtt.core.BaseMqttBinaryMessageReader;
+
+import java.util.function.BiConsumer;
+
+final class BinaryMessageReader extends BaseMqttBinaryMessageReader {
+
+ Mqtt5Publish message;
+
+ BinaryMessageReader(final SpecVersion version, final String contentType, Mqtt5Publish message) {
+ super(version, contentType, message.getPayloadAsBytes());
+
+ this.message = message;
+ }
+
+ @Override
+ protected void forEachUserProperty(BiConsumer
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.datatypes.MqttUtf8String;
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3PublishBuilder;
+import com.hivemq.client.mqtt.mqtt5.datatypes.Mqtt5UserProperty;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishBuilder;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.core.message.impl.MessageUtils;
+import io.cloudevents.core.v1.CloudEventV1;
+import io.cloudevents.mqtt.core.MqttUtils;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * A factory to obtain:
+ * - {@link MessageReader} instances to read CloudEvents from MQTT messages.
+ * - {@link MessageWriter} instances to write CloudEvents into MQTT messages.
+ */
+public class MqttMessageFactory {
+
+ // Prevent Instantiation.
+ private MqttMessageFactory() {
+ }
+
+ /**
+ * Create a {@link MessageReader} for an MQTT V3 message.
+ *
+ * As-Per MQTT Binding specification this only supports
+ * a structured JSON Format message.
+ *
+ * @param message An MQTT V3 message.
+ * @return MessageReader.
+ */
+ public static MessageReader createReader(Mqtt3Publish message) {
+ return new GenericStructuredMessageReader(MqttUtils.getDefaultEventFormat(), message.getPayloadAsBytes());
+ }
+
+ /**
+ * Create a {@link MessageReader} for an MQTT V5 message
+ *
+ * @param message An MQTT V5 message.
+ * @return A message reader.
+ */
+ public static MessageReader createReader(Mqtt5Publish message) {
+
+ Optional
+ * Only supports structured messages.
+ *
+ * @param builder {@link Mqtt3PublishBuilder.Complete}
+ * @return A message writer.
+ */
+ public static MessageWriter createWriter(Mqtt3PublishBuilder.Complete builder) {
+ return new V3MessageWriter(builder);
+ }
+
+
+ // -- Private functions
+
+ /**
+ * Find the value of the CloudEvent 'specversion' in the MQTT V5 User Properties.
+ *
+ * @param message An MQTT message.
+ * @return spec version attribute content.
+ */
+ private static String getSpecVersion(Mqtt5Publish message) {
+
+ List
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3PublishBuilder;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+
+class V3MessageWriter implements MessageWriter
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishBuilder;
+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;
+
+class V5MessageWriter implements MessageWriter
+ * Use the {@link io.cloudevents.mqtt.hivemq.MqttMessageFactory} to obtain
+ * CloudEvent reader and writer instances.
+ *
+ * Both V3 and V5 versions of MQTT are supported.
+ *
+ * @since 2.5.0
+ */
+
+package io.cloudevents.mqtt.hivemq;
diff --git a/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/MqttMessageFactoryTest.java b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/MqttMessageFactoryTest.java
new file mode 100644
index 000000000..377a24fe4
--- /dev/null
+++ b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/MqttMessageFactoryTest.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2018-Present The CloudEvents 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishBuilder;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.mock.CSVFormat;
+import io.cloudevents.core.provider.EventFormatProvider;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class MqttMessageFactoryTest {
+
+ @Test
+ public void createV3Writer() {
+ }
+
+ @Test
+ public void createV5Writer() {
+ Assertions.assertNotNull(MqttMessageFactory.createWriter((Mqtt5PublishBuilder.Complete) Mqtt5Publish.builder()));
+ }
+
+ @Test
+ public void create3Reader() {
+
+ Mqtt3Publish msg = Mqtt3Publish.builder().topic("test").build();
+ Assertions.assertNotNull(MqttMessageFactory.createReader(msg));
+ }
+
+ @Test
+ public void createV5ReaderFromStructured() {
+
+ // If the content-type is present then hopefully it's a
+ // cloudvent one.
+
+ EventFormat ef = CSVFormat.INSTANCE;
+
+ EventFormatProvider.getInstance().registerFormat(ef);
+
+ Mqtt5Publish msg = Mqtt5Publish.builder()
+ .topic("test")
+ .contentType(ef.serializedContentType())
+ .build();
+
+ Assertions.assertNotNull(MqttMessageFactory.createReader(msg));
+
+ }
+
+ @Test
+ public void createV5ReaderFromBinary() {
+
+ Mqtt5Publish msg = Mqtt5Publish.builder()
+ .topic("test")
+ .userProperties().add("specversion", "1.0").applyUserProperties()
+ .build();
+ Assertions.assertNotNull(MqttMessageFactory.createReader(msg));
+
+ }
+}
diff --git a/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3MessageWriterTest.java b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3MessageWriterTest.java
new file mode 100644
index 000000000..79897805b
--- /dev/null
+++ b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3MessageWriterTest.java
@@ -0,0 +1,71 @@
+/*
+ * Copyright 2018-Present The CloudEvents 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3PublishBuilder;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.mock.CSVFormat;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.core.test.Data;
+import io.cloudevents.rw.CloudEventRWException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class V3MessageWriterTest {
+
+ Mqtt3PublishBuilder.Complete builder;
+ V3MessageWriter writer;
+ EventFormat csvFormat = CSVFormat.INSTANCE;
+
+
+ V3MessageWriterTest() {
+
+ builder = (Mqtt3PublishBuilder.Complete) Mqtt3Publish.builder();
+ writer = new V3MessageWriter(builder);
+ EventFormatProvider.getInstance().registerFormat(csvFormat);
+ }
+
+ @Test
+ void create() {
+ }
+
+ @Test
+ void setEvent() {
+ }
+
+ @Test
+ void writeStructuredA() {
+ assertNotNull(writer.writeStructured(Data.V1_MIN, csvFormat.serializedContentType()));
+ }
+
+ @Test
+ void testWriteStructuredB() {
+ assertNotNull(writer.writeStructured(Data.V1_MIN, csvFormat));
+ }
+
+ @Test
+ void writeBinary() {
+
+ // This should fail
+ Assertions.assertThrows(CloudEventRWException.class, () -> {
+ writer.writeBinary(Data.V1_MIN);
+ });
+ }
+}
diff --git a/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3RoundTripTests.java b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3RoundTripTests.java
new file mode 100644
index 000000000..0227d4734
--- /dev/null
+++ b/bindings/mqtt/hivemq/src/test/java/io/cloudevents/mqtt/hivemq/V3RoundTripTests.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2018-Present The CloudEvents 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
+import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3PublishBuilder;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.test.Data;
+import io.cloudevents.jackson.JsonFormat;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+/**
+ * Round-Trip Tests
+ *
+ * - serialize a CloudEvent into an MQTT Message.
+ * - de-serialize the message into a new CloudEvent
+ * - verify that the new CE matches the original CE
+ */
+public class V3RoundTripTests {
+
+
+ /**
+ * This test set is limited owing to the fact that:
+ * (a) We only support JSON Format
+ * (b) Round-tripping of events with JSON 'data' doesn't reliably work owing to the way the equality tests work on the event.
+ *
+ * @return
+ */
+ static Stream
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt5.datatypes.Mqtt5UserProperty;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishBuilder;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.data.BytesCloudEventData;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class V5MessageWriterTest {
+
+ private final Mqtt5PublishBuilder builder;
+ private final V5MessageWriter writer;
+
+ V5MessageWriterTest() {
+ builder = Mqtt5Publish.builder();
+ writer = new V5MessageWriter((Mqtt5PublishBuilder.Complete) builder);
+ builder.topic("tester");
+ }
+
+ @Test
+ public void testWithContextAttribute() {
+
+ Assertions.assertNotNull(writer.withContextAttribute("test", "testing"));
+
+ Mqtt5Publish msg = ((Mqtt5PublishBuilder.Complete) builder).build();
+
+ ensureProperty(msg, "test", "testing");
+ }
+
+ @Test
+ public void testWithContextAttributes() {
+
+ Assertions.assertNotNull(writer.withContextAttribute("test1", "testing1"));
+ Assertions.assertNotNull(writer.withContextAttribute("test2", "testing2"));
+
+ Mqtt5Publish msg = ((Mqtt5PublishBuilder.Complete) builder).build();
+
+ ensureProperty(msg, "test1", "testing1");
+ ensureProperty(msg, "test2", "testing2");
+ }
+
+ @Test
+ public void testEnd() {
+ Assertions.assertNotNull(writer.end());
+ }
+
+ @Test
+ public void testEndWithData() {
+ final byte[] tData = {0x00, 0x02, 0x42};
+
+ Assertions.assertNotNull(writer.end(BytesCloudEventData.wrap(tData)));
+
+ Mqtt5Publish msg = ((Mqtt5PublishBuilder.Complete) builder).build();
+
+ Assertions.assertNotNull(msg.getPayloadAsBytes());
+ Assertions.assertEquals(msg.getPayloadAsBytes().length, tData.length);
+
+ }
+
+ @Test
+ public void testCreate() {
+ Assertions.assertNotNull(writer.create(SpecVersion.V1));
+
+ Mqtt5Publish msg = ((Mqtt5PublishBuilder.Complete) builder).build();
+ ensureProperty(msg, "specversion", SpecVersion.V1.toString());
+
+ }
+
+ private void ensureProperty(Mqtt5Publish msg, String name, String val) {
+
+ List
+ * 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
+ *
+ * http://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.mqtt.hivemq;
+
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
+import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishBuilder;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.mock.CSVFormat;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Round-Trip Tests
+ *
+ * For both Binary and Structured modes:
+ * - serialize a CloudEvent into an MQTT Message.
+ * - de-serialize the message into a new CloudEvent
+ * - verify that the new CE matches the original CE
+ */
+public class V5RoundTripTests {
+
+ private static void readAndVerify(CloudEvent ce, Mqtt5Publish message) {
+
+ Assertions.assertNotNull(message);
+
+ // Read the message back into an event
+ MessageReader reader = MqttMessageFactory.createReader(message);
+ Assertions.assertNotNull(reader);
+
+ CloudEvent newCE = reader.toEvent();
+ Assertions.assertNotNull(newCE);
+
+ // And now ensure we got back what we wrote
+ Assertions.assertEquals(ce, newCE);
+ }
+
+ @ParameterizedTest
+ @MethodSource("io.cloudevents.core.test.Data#allEventsWithoutExtensions")
+ public void roundTripBinary(CloudEvent ce) {
+
+ // Write the event out as a message.
+ Mqtt5Publish message = null;
+ Mqtt5PublishBuilder.Complete builder = (Mqtt5PublishBuilder.Complete) Mqtt5Publish.builder();
+ builder.topic("test.test.test");
+
+
+ MessageWriter writer = MqttMessageFactory.createWriter(builder);
+ Assertions.assertNotNull(writer);
+
+ writer.writeBinary(ce);
+
+ message = builder.build();
+
+ // Read it back and verify
+ readAndVerify(ce, message);
+ }
+
+ @ParameterizedTest
+ @MethodSource("io.cloudevents.core.test.Data#allEventsWithoutExtensions")
+ public void roundTripStructured(CloudEvent ce) {
+
+ EventFormat format = CSVFormat.INSTANCE;
+
+ Mqtt5Publish message = null;
+ Mqtt5PublishBuilder.Complete builder = (Mqtt5PublishBuilder.Complete) Mqtt5Publish.builder();
+ builder.topic("test.test.test");
+
+ // Write the event out as a message.
+ MessageWriter writer = MqttMessageFactory.createWriter(builder);
+ Assertions.assertNotNull(writer);
+
+ writer.writeStructured(ce, format);
+
+ message = builder.build();
+
+ // Read it back and verify
+ readAndVerify(ce, message);
+
+ }
+
+
+}
diff --git a/bindings/mqtt/paho/pom.xml b/bindings/mqtt/paho/pom.xml
new file mode 100644
index 000000000..7324163f8
--- /dev/null
+++ b/bindings/mqtt/paho/pom.xml
@@ -0,0 +1,106 @@
+
+
+
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.SpecVersion;
+import io.cloudevents.mqtt.core.BaseMqttBinaryMessageReader;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.UserProperty;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+final class BinaryMessageReader extends BaseMqttBinaryMessageReader {
+
+ private final List
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.packet.UserProperty;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * General Utility functions
+ */
+final class PahoMessageUtils {
+
+ /**
+ * Prevent Instantiation
+ */
+ private PahoMessageUtils() {
+ }
+
+ /**
+ * Get the value of a specific user property from a message.
+ *
+ * @param msg The MQTT Message
+ * @param name The property to retrieve.
+ * @return property value or NULL if not set.
+ */
+ static String getUserProperty(final MqttMessage msg, final String name) {
+
+ final MqttProperties mProps = msg.getProperties();
+
+ return (mProps == null) ? null : getUserProperty(mProps.getUserProperties(), name);
+
+ }
+
+ /**
+ * Get the value of a specific user property from a message.
+ *
+ * @param props The List of MQTT Message properties
+ * @param name The property to retrieve.
+ * @return property value or NULL if not set.
+ */
+ public static String getUserProperty(final List
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.provider.EventFormatProvider;
+import io.cloudevents.mqtt.core.MqttUtils;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+/**
+ * A {@link MessageWriter} that writes an CloudEvent to a V3 MQTT Message.
+ *
+ * Note: This only supports Structured messages in JSON format as defined
+ * by the MQTT CloudEvent binding specification.
+ */
+class V3MessageWriter implements MessageWriter
+ * Raises exception if not valid.
+ *
+ * @param contentType
+ */
+ private void ensureValidContent(String contentType) {
+
+ if (!MqttUtils.getDefaultContentType().equals(contentType)) {
+
+ throw CloudEventRWException.newOther("MQTT V3 Does not support contentType: " + contentType);
+
+ }
+ }
+
+ @Override
+ public MqttMessage writeStructured(CloudEvent event, String format) {
+
+ final EventFormat eventFormat = EventFormatProvider.getInstance().resolveFormat(format);
+
+ // Sanity Check
+ if (eventFormat == null) {
+
+ }
+
+ return writeStructured(event, eventFormat);
+ }
+
+ @Override
+ public MqttMessage writeStructured(CloudEvent event, EventFormat format) {
+ // Ensure format is valid
+ ensureValidContent(format.serializedContentType());
+ // Populate the structured format.
+ message.setPayload(format.serialize(event));
+ // Done.
+ return message;
+ }
+
+ @Override
+ public MqttMessage writeBinary(CloudEvent event) {
+ // This operation is not allowed.
+ // This should fail
+ throw CloudEventRWException.newOther("MQTT V3 Does not support CloudEvent Binary mode");
+ }
+
+ @Override
+ public CloudEventWriter
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.mqtt.core.MqttUtils;
+import io.cloudevents.rw.CloudEventWriter;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+/**
+ * MQTT V3 factory to :
+ * - Obtain a {@link MessageReader} to read CloudEvents from MQTT messages.
+ * - Create a {@link MessageWriter} enabling CloudEVents to be written to an MQTT message.
+ *
+ * NOTE: The V3 binding only supports structured messages using a JSON Format.
+ */
+
+public final class V3MqttMessageFactory {
+
+ /**
+ * Prevent instantiation.
+ */
+ private V3MqttMessageFactory() {
+
+ }
+
+ /**
+ * Create a {@link MessageReader} to read a V3 MQTT Messages as a CloudEVents
+ *
+ * @param mqttMessage An MQTT Message.
+ * @return {@link MessageReader}
+ */
+ public static MessageReader createReader(MqttMessage mqttMessage) {
+ return new GenericStructuredMessageReader(MqttUtils.getDefaultEventFormat(), mqttMessage.getPayload());
+ }
+
+ /**
+ * Creates a {@link MessageWriter} to write a CloudEvent to an MQTT {@link MqttMessage}.
+ *
+ * NOTE: This implementation *only* supports JSON structured format as-per the MQTT binding specification.
+ *
+ * @return A {@link MessageWriter} to write a {@link io.cloudevents.CloudEvent} to MQTT.
+ */
+ public static MessageWriter
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+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.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.packet.UserProperty;
+
+import java.util.ArrayList;
+import java.util.List;
+
+class V5MessageWriter
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.core.message.impl.MessageUtils;
+import io.cloudevents.core.v1.CloudEventV1;
+import io.cloudevents.rw.CloudEventWriter;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+
+/**
+ * MQTT V5 factory to :
+ * - Obtain a {@link MessageReader} to read CloudEvents from MQTT messages.
+ * - Create a {@link MessageWriter} enabling CloudEVents to be written to an MQTT message.
+ */
+
+public final class V5MqttMessageFactory {
+
+ /**
+ * Prevent instantiation.
+ */
+ private V5MqttMessageFactory() {
+
+ }
+
+ /**
+ * Create a {@link MessageReader} to read MQTT Messages as CloudEVents
+ *
+ * @param mqttMessage An MQTT Message.
+ * @return {@link MessageReader}
+ */
+ public static MessageReader createReader(MqttMessage mqttMessage) {
+
+ final String contentType = mqttMessage.getProperties().getContentType();
+
+ return MessageUtils.parseStructuredOrBinaryMessage(
+ () -> contentType,
+ format -> new GenericStructuredMessageReader(format, mqttMessage.getPayload()),
+ () -> PahoMessageUtils.getUserProperty(mqttMessage, CloudEventV1.SPECVERSION),
+ sv -> new BinaryMessageReader(sv, contentType, mqttMessage)
+ );
+ }
+
+ /**
+ * Creates a {@link MessageWriter} capable of translating both a structured and binary CloudEvent
+ * to an MQTT {@link MqttMessage}
+ *
+ * @return A {@link MessageWriter} to write a {@link io.cloudevents.CloudEvent} to MQTT using structured or binary encoding.
+ */
+ public static MessageWriter
+ * Separate factories are provided for MQTT V3 and V5.
+ *
+ * @since 2.5.0
+ */
+
+package io.cloudevents.mqtt.paho;
diff --git a/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/PahoMessageUtilsTest.java b/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/PahoMessageUtilsTest.java
new file mode 100644
index 000000000..a7661c306
--- /dev/null
+++ b/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/PahoMessageUtilsTest.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2018-Present The CloudEvents 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
+ *
+ * http://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.mqtt.paho;
+
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.packet.UserProperty;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class PahoMessageUtilsTest {
+
+ @Test
+ void verifyPropertyList() {
+
+ List
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventDeserializationException;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.mock.CSVFormat;
+import io.cloudevents.core.test.Data;
+import io.cloudevents.rw.CloudEventRWException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class V3MessageFactoryTest {
+
+ @Test
+ public void ensureSerializationFormat() {
+
+ MqttMessage message = null;
+
+ // This should fail as we don't support CSV Format
+
+ MessageWriter writer = V3MqttMessageFactory.createWriter();
+ Assertions.assertNotNull(writer);
+
+ // Expect an exception
+
+ Assertions.assertThrows(CloudEventRWException.class, () -> {
+ writer.writeStructured(Data.V1_MIN, CSVFormat.INSTANCE);
+ });
+ }
+
+
+ @ParameterizedTest()
+ @MethodSource("io.cloudevents.core.test.Data#allEventsWithoutExtensions")
+ public void ensureDeserialization(CloudEvent ce) {
+
+
+ final String contentType = CSVFormat.INSTANCE.serializedContentType() + "; charset=utf8";
+ final byte[] contentPayload = CSVFormat.INSTANCE.serialize(ce);
+
+ // Build the MQTT Message
+
+ MqttMessage m = new MqttMessage();
+ m.setPayload(contentPayload);
+
+ // Get a reader
+ MessageReader reader = V3MqttMessageFactory.createReader(m);
+ Assertions.assertNotNull(reader);
+
+ // This should fail
+ // Expect an exception
+
+ Assertions.assertThrows(EventDeserializationException.class, () -> {
+ reader.toEvent();
+ });
+
+ }
+}
diff --git a/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/V3RoundTripTests.java b/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/V3RoundTripTests.java
new file mode 100644
index 000000000..6be040f61
--- /dev/null
+++ b/bindings/mqtt/paho/src/test/java/io/cloudevents/mqtt/paho/V3RoundTripTests.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2018-Present The CloudEvents 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.test.Data;
+import io.cloudevents.jackson.JsonFormat;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+/**
+ * Round-Trip Tests
+ *
+ * - serialize a CloudEvent into an MQTT Message.
+ * - de-serialize the message into a new CloudEvent
+ * - verify that the new CE matches the original CE
+ */
+public class V3RoundTripTests {
+
+
+ /**
+ * This test set is limited owing to the the fact that:
+ * (a) We only support JSON Format
+ * (b) Round-tripping of events with JSON 'data' doesn't reliably work owing to the way the equality tests work on the event.
+ *
+ * @return
+ */
+ static Stream
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.message.Encoding;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.mock.CSVFormat;
+import io.cloudevents.core.test.Data;
+import io.cloudevents.core.v03.CloudEventV03;
+import io.cloudevents.types.Time;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.packet.UserProperty;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class V5MessageFactoryTest {
+
+ private static final String DATACONTENTTYPE_NULL = null;
+ private static final byte[] DATAPAYLOAD_NULL = null;
+
+ private static Stream
+ * 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
+ *
+ * http://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.mqtt.paho;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.mock.CSVFormat;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Round-Trip Tests
+ *
+ * For both Binary and Structured modes:
+ * - serialize a CloudEvent into an MQTT Message.
+ * - de-serialize the message into a new CloudEvent
+ * - verify that the new CE matches the original CE
+ */
+public class V5RoundTripTests {
+
+ private static void readAndVerify(CloudEvent ce, MqttMessage message) {
+
+ Assertions.assertNotNull(message);
+
+ // Read the message back into an event
+ MessageReader reader = V5MqttMessageFactory.createReader(message);
+ Assertions.assertNotNull(reader);
+
+ CloudEvent newCE = reader.toEvent();
+ Assertions.assertNotNull(newCE);
+
+ // And now ensure we got back what we wrote
+ Assertions.assertEquals(ce, newCE);
+ }
+
+ @ParameterizedTest
+ @MethodSource("io.cloudevents.core.test.Data#allEventsWithoutExtensions")
+ public void roundTripBinary(CloudEvent ce) {
+
+ // Write the event out as a message.
+ MessageWriter writer = V5MqttMessageFactory.createWriter();
+ Assertions.assertNotNull(writer);
+
+ MqttMessage message = (MqttMessage) writer.writeBinary(ce);
+
+ // Read it back and verify
+ readAndVerify(ce, message);
+ }
+
+ @ParameterizedTest
+ @MethodSource("io.cloudevents.core.test.Data#allEventsWithoutExtensions")
+ public void roundTripStructured(CloudEvent ce) {
+
+ EventFormat format = CSVFormat.INSTANCE;
+
+ // Write the event out as a message.
+ MessageWriter writer = V5MqttMessageFactory.createWriter();
+ Assertions.assertNotNull(writer);
+
+ MqttMessage message = (MqttMessage) writer.writeStructured(ce, format);
+
+ // Read it back and verify
+ readAndVerify(ce, message);
+
+ }
+
+
+}
diff --git a/docs/index.md b/docs/index.md
index d8ec2ebe9..8078c4e3d 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -27,23 +27,26 @@ Using the Java SDK you can:
## Supported features
| | [v0.3](https://github.com/cloudevents/spec/tree/v0.3) | [v1.0](https://github.com/cloudevents/spec/tree/v1.0) |
-| :------------------------------------------------: | :---------------------------------------------------: | :---------------------------------------------------: |
+|:--------------------------------------------------:|:-----------------------------------------------------:|:-----------------------------------------------------:|
| CloudEvents Core | :heavy_check_mark: | :heavy_check_mark: |
| AMQP Protocol Binding | :x: | :x: |
| - [Proton](amqp-proton.md) | :heavy_check_mark: | :heavy_check_mark: |
+ | MQTT Protocol Binding | :x: | :x: |
+ | - [Paho](mqtt.md) | :heavy_check_mark: | :heavy_check_mark: |
+ | - [HiveMQ](mqtt.md) | :heavy_check_mark: | :heavy_check_mark: |
| AVRO Event Format | :x: | :x: |
| HTTP Protocol Binding | :heavy_check_mark: | :heavy_check_mark: |
| - [Vert.x](http-vertx.md) | :heavy_check_mark: | :heavy_check_mark: |
| - [Jakarta Restful WS](http-jakarta-restful-ws.md) | :heavy_check_mark: | :heavy_check_mark: |
| - [Basic](http-basic.md) | :heavy_check_mark: | :heavy_check_mark: |
| - [Spring](spring.md) | :heavy_check_mark: | :heavy_check_mark: |
-| - [http4k][http4k]† | :heavy_check_mark: | :heavy_check_mark: |
+| - [http4k][http4k]† | :heavy_check_mark: | :heavy_check_mark: |
| JSON Event Format | :heavy_check_mark: | :heavy_check_mark: |
| - [Jackson](json-jackson.md) | :heavy_check_mark: | :heavy_check_mark: |
-| Protobuf Event Format | :heavy_check_mark: | :heavy_check_mark: |
-| - [Proto](protobuf.md) | :heavy_check_mark: | :heavy_check_mark: |
-| XML Event Format | :heavy_check_mark: | :heavy_check_mark: |
-| - [XML](xml.md) | :heavy_check_mark: | :heavy_check_mark: |
+| Protobuf Event Format | :heavy_check_mark: | :heavy_check_mark: |
+| - [Proto](protobuf.md) | :heavy_check_mark: | :heavy_check_mark: |
+| XML Event Format | :heavy_check_mark: | :heavy_check_mark: |
+| - [XML](xml.md) | :heavy_check_mark: | :heavy_check_mark: |
| [Kafka Protocol Binding](kafka.md) | :heavy_check_mark: | :heavy_check_mark: |
| MQTT Protocol Binding | :x: | :x: |
| NATS Protocol Binding | :x: | :x: |
diff --git a/docs/mqtt.md b/docs/mqtt.md
new file mode 100644
index 000000000..4d1aacee0
--- /dev/null
+++ b/docs/mqtt.md
@@ -0,0 +1,60 @@
+---
+title: CloudEvents MQTT
+nav_order: 5
+---
+
+# MQTT Support
+
+The SDK supports both V3 and V5 MQTT binding specifications via these Java client libraries:
+
+ * [Paho]()
+ * [HiveMQ]()
+
+NOTE: MQTT V3 *only* supports structured mode transfer of CloudEVents. Operations related to binary mode transmission
+are either not available or will throw runtime exceptions if an attempt is made to use them.
+
+Both client library implementations rely on a *provided* maven dependency.
+
+# General Usage
+
+There is a slight variance in usage between the two supported client libraries owing to the way those clients
+have implemented support for the two versions of MQTT but the general pattern is the same as every other protocol
+binding.
+
+## Creating a message from a CloudEvent
+
+ 1. Obtain a `MessageWriter` from a factory.
+ 2. Use the writer to populate the MQTT message using structured or binary mode.
+ * `mqttMessage = messageWriter.writeBinary(cloudEvent);` or,
+ * `mqttMessage = messageWriter.writeStructured(cloudEvent, eventFormat);`
+
+## Creating a CloudEvent from a message.
+
+ 1. Obtain a 'MessageReader' from a message factory for an MQTT message.
+ 2. Obtain a CloudEvent from the reader.
+ * _CloudEvent cloudEvent = reader.toEvent();_
+
+
+# PAHO Client Usage
+
+## Maven
+
+```xml
+