Merge pull request #2804 from particle-iot/typed-publish/sc-129495

Add support for event content types

communication/src/coap_defs.h → communication/inc/coap_defs.h

@@ -108,13 +108,19 @@ enum class CoapOption {
 PARTICLE_DEFINE_ENUM_COMPARISON_OPERATORS(CoapOption)
 
 enum class CoapContentFormat {
-    // RFC 7252 12.3. CoAP Content-Formats Registry
-    TEXT_PLAIN_CHARSET_UTF8 = 0,
+    // RFC 7252
+    TEXT_PLAIN = 0, // text/plain;charset=utf-8
     APPLICATION_LINK_FORMAT = 40,
     APPLICATION_XML = 41,
     APPLICATION_OCTET_STREAM = 42,
     APPLICATION_EXI = 47,
-    APPLICATION_JSON = 50
+    APPLICATION_JSON = 50,
+    // https://www.iana.org/assignments/core-parameters/core-parameters.xhtml#content-formats
+    IMAGE_JPEG = 22,
+    IMAGE_PNG = 23,
+    APPLICATION_CBOR = 60,
+    // Vendor-specific formats
+    PARTICLE_JSON_AS_CBOR = 65001 // application/vnd.particle.json+cbor
 };
 
 PARTICLE_DEFINE_ENUM_COMPARISON_OPERATORS(CoapContentFormat)
@@ -188,6 +194,8 @@ inline bool isCoapEmptyAck(CoapType type, unsigned code) {
     return type == CoapType::ACK && code == CoapCode::EMPTY;
 }
 
+bool isCoapTextContentFormat(unsigned fmt);
+
 } // namespace protocol
 
 } // namespace particle

communication/inc/coap_util.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include "coap_api.h"
+#include "coap_defs.h"
 #include "logging.h"
 
 namespace particle {
@@ -86,6 +87,37 @@ class CoapMessagePtr {
 
 namespace protocol {
 
+class Message;
+class MessageChannel;
+class CoapOptionIterator;
+
+/**
+ * Send an empty ACK or RST message.
+ *
+ * @param channel Message channel.
+ * @param msg Received message.
+ * @param type Type of the message to send.
+ * @return 0 on success, otherwise an error code defined by the `system_error_t` enum.
+ */
+int sendEmptyAckOrRst(MessageChannel& channel, Message& msg, CoapType type);
+
+/**
+ * Append an URI path entry to a string.
+ *
+ * If the path in the buffer is not empty (`pathLen > 0`), appends a separator character to it,
+ * otherwise appends just the value of the URI path option.
+ *
+ * The output is null-terminated unless the size of the buffer is 0.
+ *
+ * @param buf Destination buffer.
+ * @param bufSize Buffer size.
+ * @param pathLen Length of the URI path already stored in the buffer.
+ * @param it Iterator pointing to a URI path CoAP option.
+ * @return Length of the URI path entry plus one character for a path separator if the path in the
+ *        buffer wasn't empty.
+ */
+size_t appendUriPath(char* buf, size_t bufSize, size_t pathLen, const CoapOptionIterator& it);
+
 /**
  * Log the contents of a CoAP message.
  *

communication/inc/communication_dynalib.h

@@ -44,9 +44,9 @@ DYNALIB_FN(7, communication, spark_protocol_event_loop, bool(ProtocolFacade* pro
 DYNALIB_FN(8, communication, spark_protocol_is_initialized, bool(ProtocolFacade*))
 DYNALIB_FN(9, communication, spark_protocol_presence_announcement, int(ProtocolFacade*, uint8_t*, const uint8_t*, void*))
 DYNALIB_FN(10, communication, spark_protocol_send_event, bool(ProtocolFacade*, const char*, const char*, int, uint32_t, void*))
-DYNALIB_FN(11, communication, spark_protocol_send_subscription_device, bool(ProtocolFacade*, const char*, const char*, void*))
-DYNALIB_FN(12, communication, spark_protocol_send_subscription_scope, bool(ProtocolFacade*, const char*, SubscriptionScope::Enum, void*))
-DYNALIB_FN(13, communication, spark_protocol_add_event_handler, bool(ProtocolFacade*, const char*, EventHandler, SubscriptionScope::Enum, const char*, void*))
+DYNALIB_FN(11, communication, spark_protocol_send_subscription_device_deprecated, bool(ProtocolFacade*, const char*, const char*, void*))
+DYNALIB_FN(12, communication, spark_protocol_send_subscription, bool(ProtocolFacade*, const char*, uint8_t, void*))
+DYNALIB_FN(13, communication, spark_protocol_add_event_handler, bool(ProtocolFacade*, const char*, EventHandler, uint8_t, const char*, void*))
 DYNALIB_FN(14, communication, spark_protocol_send_time_request, bool(ProtocolFacade*, void*))
 DYNALIB_FN(15, communication, spark_protocol_send_subscriptions, void(ProtocolFacade*, void*))
 

communication/inc/events.h

@@ -28,22 +28,23 @@
 #include <stdint.h>
 #include <stdlib.h>
 #include "platforms.h"
+#include "protocol_defs.h"
 
 namespace EventType {
   enum Enum : char {
-    PUBLIC = 'e',			// 0x65
-    PRIVATE = 'E',			// 0x45
+    PUBLIC = 'e', // Deprecated (0x65)
+    PRIVATE = 'E', // Deprecated (0x45)
   };
 
   /**
    * These flags are encoded into the same 32-bit integer that already holds EventType::Enum
    */
   enum Flags {
-	  EMPTY_FLAGS = 0,
-	   NO_ACK = 0x2,
-	   WITH_ACK = 0x8,
-	   ASYNC = 0x10,        // not used here, but reserved since it's used in the system layer. Makes conversion simpler.
-	   ALL_FLAGS = NO_ACK | WITH_ACK | ASYNC
+    EMPTY_FLAGS = 0,
+    NO_ACK = 0x2,
+    WITH_ACK = 0x8,
+    ASYNC = 0x10,        // not used here, but reserved since it's used in the system layer. Makes conversion simpler.
+    ALL_FLAGS = NO_ACK | WITH_ACK | ASYNC
   };
 
   static_assert((PUBLIC & NO_ACK)==0 &&
@@ -52,51 +53,39 @@ namespace EventType {
 	  (PRIVATE & WITH_ACK)==0 &&
 	  (PRIVATE & ASYNC)==0 &&
 	  (PUBLIC & ASYNC)==0, "flags should be distinct from event type");
-
-/**
- * The flags are encoded in with the event type.
- */
-  inline Enum extract_event_type(uint32_t& value)
-  {
-	  Enum et = Enum(value & ~ALL_FLAGS);
-	  value = value & ALL_FLAGS;
-	  return et;
-  }
 } // namespace EventType
 
 #if PLATFORM_ID != PLATFORM_GCC
-static_assert(sizeof(EventType::Enum)==1, "EventType size is 1");
+static_assert(sizeof(EventType::Enum) == 1, "EventType::Enum size is not 1");
 #endif
 
-namespace SubscriptionScope {
+namespace SubscriptionFlag {
   enum Enum {
-    MY_DEVICES,
-    FIREHOSE
+    MY_DEVICES = 0x00, // Deprecated
+    FIREHOSE = 0x01, // Deprecated
+    BINARY_DATA = 0x02, // The subscription handler accepts binary data
+    CBOR_DATA = 0x04 // The subscription handler accepts CBOR data
   };
 }
 
+#if PLATFORM_ID != PLATFORM_GCC
+static_assert(sizeof(SubscriptionFlag::Enum) == 1, "SubscriptionFlag::Enum size is not 1");
+#endif
+
 typedef void (*EventHandler)(const char *event_name, const char *data);
-typedef void (*EventHandlerWithData)(void *handler_data, const char *event_name, const char *data);
+typedef void (*EventHandlerWithData)(void *handler_data, const char *event_name, const char *data, size_t data_size,
+    int content_type);
 
 /**
  *  This is used in a callback so only change by adding fields to the end
  */
 struct FilteringEventHandler
 {
-  char filter[64];
+  char filter[64]; // XXX: Not null-terminated if 64 characters long
   EventHandler handler;
   void *handler_data;
-  SubscriptionScope::Enum scope;
-  char device_id[13];
+  uint8_t flags;
+  char device_id[13]; // XXX: Unused field. Keeping for ABI compatibility for now
 };
 
-
-size_t subscription(uint8_t buf[], uint16_t message_id,
-                    const char *event_name, const char *device_id);
-
-size_t subscription(uint8_t buf[], uint16_t message_id,
-                    const char *event_name, SubscriptionScope::Enum scope);
-
-size_t event_name_uri_path(uint8_t buf[], const char *name, size_t name_len);
-
 #endif // __EVENTS_H

communication/inc/messages.h

@@ -79,10 +79,6 @@ class Messages
 			unsigned char token, unsigned char code, const unsigned char* payload,
 			unsigned payload_len, bool confirmable);
 
-	static size_t event(uint8_t buf[], uint16_t message_id, const char *event_name,
-	             const char *data, size_t data_size, int ttl, EventType::Enum event_type, bool confirmable);
-
-
     static inline size_t empty_ack(unsigned char *buf,
                           unsigned char message_id_msb,
                           unsigned char message_id_lsb) {

communication/inc/protocol.h

@@ -500,8 +500,8 @@ class Protocol
 	ProtocolError post_description(int desc_flags, bool force);
 
 	// Returns true on success, false on sending timeout or rate-limiting failure
-	bool send_event(const char *event_name, const char *data, int ttl,
-			EventType::Enum event_type, int flags, CompletionHandler handler)
+	bool send_event(const char *event_name, const char *data, size_t data_size, int content_type, int ttl, int flags,
+			CompletionHandler handler)
 	{
 #if HAL_PLATFORM_OTA_PROTOCOL_V3
 		const bool updating = firmwareUpdate.isRunning();
@@ -513,8 +513,8 @@ class Protocol
 			handler.setError(SYSTEM_ERROR_BUSY);
 			return false;
 		}
-		const ProtocolError error = publisher.send_event(channel, event_name, data, ttl, event_type, flags,
-				callbacks.millis(), std::move(handler));
+		const ProtocolError error = publisher.send_event(channel, event_name, data, data_size, content_type, ttl,
+				flags, callbacks.millis(), std::move(handler));
 		if (error != NO_ERROR)
 		{
 			handler.setError(toSystemError(error));
@@ -525,28 +525,24 @@ class Protocol
 
 	void build_describe_message(Appender& appender, int desc_flags);
 
-	inline bool add_event_handler(const char *event_name, EventHandler handler)
+	bool add_event_handler(const char *event_name, EventHandler handler)
 	{
-		return add_event_handler(event_name, handler, NULL,
-				SubscriptionScope::FIREHOSE, NULL);
+		return add_event_handler(event_name, handler, nullptr /* handler_data */, 0 /* flags */);
 	}
 
-	inline bool add_event_handler(const char *event_name, EventHandler handler,
-			void *handler_data, SubscriptionScope::Enum scope,
-			const char* device_id)
+	bool add_event_handler(const char *event_name, EventHandler handler, void *handler_data, int flags)
 	{
-		return !subscriptions.add_event_handler(event_name, handler,
-				handler_data, scope, device_id);
+		auto err = subscriptions.add_event_handler(event_name, handler, handler_data, flags);
+		return err == ProtocolError::NO_ERROR;
 	}
 
-	inline bool remove_event_handlers(const char* name)
+	bool remove_event_handlers(const char* name)
 	{
 		subscriptions.remove_event_handlers(name);
 		return true;
 	}
 
-	ProtocolError send_subscription(const char *event_name, const char *device_id);
-	ProtocolError send_subscription(const char *event_name, SubscriptionScope::Enum scope);
+	ProtocolError send_subscription(const char *event_name, int flags);
 	ProtocolError send_subscriptions(bool force);
 
 	/**

communication/inc/spark_descriptor.h

@@ -68,6 +68,9 @@ namespace SparkAppStateUpdate {
 
 struct SparkDescriptor
 {
+    typedef void (*CallEventHandlerCallback)(uint16_t size, FilteringEventHandler* handler, const char* event,
+        const char* data, size_t data_size, int content_type);
+
     typedef std::function<bool(const void*, SparkReturnType::Enum)> FunctionResultCallback;
 
     /**
@@ -96,7 +99,7 @@ struct SparkDescriptor
 
     bool (*append_system_info)(appender_fn appender, void* append, void* reserved);
 
-    void (*call_event_handler)(uint16_t size, FilteringEventHandler* handler, const char* event, const char* data, void* reserved);
+    CallEventHandlerCallback call_event_handler;
 
     /**
      * Optional callback - may be null.

communication/inc/spark_protocol_functions.h

@@ -191,15 +191,17 @@ typedef struct {
     size_t size;
     completion_callback handler_callback;
     void* handler_data;
+    size_t data_size;
+    int content_type;
 } completion_handler_data;
 
 typedef completion_handler_data spark_protocol_send_event_data;
 
 bool spark_protocol_send_event(ProtocolFacade* protocol, const char *event_name, const char *data,
                 int ttl, uint32_t flags, void* reserved);
-bool spark_protocol_send_subscription_device(ProtocolFacade* protocol, const char *event_name, const char *device_id, void* reserved=NULL);
-bool spark_protocol_send_subscription_scope(ProtocolFacade* protocol, const char *event_name, SubscriptionScope::Enum scope, void* reserved=NULL);
-bool spark_protocol_add_event_handler(ProtocolFacade* protocol, const char *event_name, EventHandler handler, SubscriptionScope::Enum scope, const char* id, void* handler_data=NULL);
+bool spark_protocol_send_subscription_device_deprecated(ProtocolFacade* protocol, const char *event_name, const char *device_id, void* reserved=NULL);
+bool spark_protocol_send_subscription(ProtocolFacade* protocol, const char *event_name, uint8_t flags, void* reserved=NULL);
+bool spark_protocol_add_event_handler(ProtocolFacade* protocol, const char *event_name, EventHandler handler, uint8_t flags, const char* device_id_deprecated, void* handler_data=NULL);
 bool spark_protocol_send_time_request(ProtocolFacade* protocol, void* reserved=NULL);
 void spark_protocol_send_subscriptions(ProtocolFacade* protocol, void* reserved=NULL);
 void spark_protocol_remove_event_handlers(ProtocolFacade* protocol, const char *event_name, void* reserved=NULL);

communication/src/build.mk

@@ -14,7 +14,6 @@ INCLUDE_DIRS += $(TARGET_SRC_PATH)
 # C++ source files included in this build.
 CPPSRC += $(TARGET_SRC_PATH)/coap.cpp
 CPPSRC += $(TARGET_SRC_PATH)/handshake.cpp
-CPPSRC += $(TARGET_SRC_PATH)/events.cpp
 CPPSRC += $(TARGET_SRC_PATH)/spark_protocol_functions.cpp
 CPPSRC += $(TARGET_SRC_PATH)/communication_dynalib.cpp
 CPPSRC += $(TARGET_SRC_PATH)/dsakeygen.cpp
@@ -28,6 +27,7 @@ CPPSRC += $(TARGET_SRC_PATH)/messages.cpp
 CPPSRC += $(TARGET_SRC_PATH)/chunked_transfer.cpp
 CPPSRC += $(TARGET_SRC_PATH)/coap_channel.cpp
 CPPSRC += $(TARGET_SRC_PATH)/publisher.cpp
+CPPSRC += $(TARGET_SRC_PATH)/subscriptions.cpp
 CPPSRC += $(TARGET_SRC_PATH)/protocol_defs.cpp
 CPPSRC += $(TARGET_SRC_PATH)/protocol_util.cpp
 CPPSRC += $(TARGET_SRC_PATH)/communication_diagnostic.cpp

communication/src/coap_defs.cpp

@@ -60,6 +60,18 @@ CoapCode coapCodeForSystemError(int error) {
     }
 }
 
+bool isCoapTextContentFormat(unsigned fmt) {
+    switch (fmt) {
+    case (unsigned)CoapContentFormat::TEXT_PLAIN:
+    case (unsigned)CoapContentFormat::APPLICATION_LINK_FORMAT:
+    case (unsigned)CoapContentFormat::APPLICATION_XML:
+    case (unsigned)CoapContentFormat::APPLICATION_JSON:
+        return true;
+    default:
+        return false;
+    }
+}
+
 } // namespace protocol
 
 } // namespace particle