Merge pull request #1486 from garlick/event_api

libflux: add flux_event_encode_raw(), flux_event_decode_raw()

doc/man3/Makefile.am

@@ -104,6 +104,8 @@ MAN3_FILES_SECONDARY = \
 	flux_event_unpack.3 \
 	flux_event_encode.3 \
 	flux_event_pack.3 \
+	flux_event_encode_raw.3 \
+	flux_event_decode_raw.3 \
 	flux_response_decode_raw.3 \
 	flux_request_encode_raw.3 \
 	flux_content_load_get.3 \
@@ -142,7 +144,6 @@ MAN3_FILES_SECONDARY = \
 	flux_kvs_txn_unlink.3 \
 	flux_kvs_txn_symlink.3 \
 	flux_kvs_txn_put_raw.3 \
-	# flux_kvs_txn_put_treeobj.3 \
 	flux_kvs_namespace_remove.3 \
 	flux_kvs_namespace_itr_next.3 \
 	flux_kvs_namespace_itr_rewind.3 \
@@ -223,6 +224,8 @@ flux_request_decode_raw.3: flux_request_decode.3
 flux_event_unpack.3: flux_event_decode.3
 flux_event_encode.3: flux_event_decode.3
 flux_event_pack.3: flux_event_decode.3
+flux_event_encode_raw.3: flux_event_decode.3
+flux_event_decode_raw.3: flux_event_decode.3
 flux_response_decode_raw.3: flux_response_decode.3
 flux_request_encode_raw.3: flux_request_encode.3
 flux_content_load_get.3: flux_content_load.3
@@ -261,7 +264,7 @@ flux_kvs_txn_mkdir.3: flux_kvs_txn_create.3
 flux_kvs_txn_unlink.3: flux_kvs_txn_create.3
 flux_kvs_txn_symlink.3: flux_kvs_txn_create.3
 flux_kvs_txn_put_raw.3: flux_kvs_txn_create.3
-# flux_kvs_txn_put_treeobj.3: flux_kvs_txn_create.3
+# N.B. exceeds max 8 stubs  flux_kvs_txn_put_treeobj.3: flux_kvs_txn_create.3
 flux_kvs_namespace_remove.3: flux_kvs_namespace_create.3
 flux_kvs_namespace_itr_next.3: flux_kvs_namespace_list.3
 flux_kvs_namespace_itr_rewind.3: flux_kvs_namespace_list.3

doc/man3/flux_event_decode.adoc

@@ -5,7 +5,7 @@ flux_event_decode(3)
 
 NAME
 ----
-flux_event_decode, flux_event_unpack, flux_event_encode, flux_event_pack - encode/decode a Flux event message
+flux_event_decode, flux_event_decode_raw, flux_event_unpack, flux_event_encode, flux_event_encode_raw, flux_event_pack - encode/decode a Flux event message
 
 
 SYNOPSIS
@@ -16,13 +16,20 @@ SYNOPSIS
                         const char **topic,
                         const char **json_str);
 
+ int flux_event_decode_raw (const flux_msg_t *msg,
+                            const char **topic,
+                            const void **data, int *len);
+
  int flux_event_unpack (const flux_msg_t *msg,
                         const char **topic,
                         const char *fmt, ...);
 
  flux_msg_t *flux_event_encode (const char *topic,
                                 const char *json_str);
 
+ flux_msg_t *flux_event_encode_raw (const char *topic,
+                                    const void *data, int len);
+
  flux_msg_t *flux_event_pack (const char *topic,
                               const char *fmt, ...);
 
@@ -39,6 +46,10 @@ _json_str_, if non-NULL, will be set to the message's JSON payload. If
 no payload exists, _json_str_ is set to NULL.  The storage for this
 string belongs to _msg_ and should not be freed.
 
+`flux_event_decode_raw()` decodes an event message with a raw payload,
+setting _data_ and _len_ to the payload data and length.  The storage for
+the raw payload belongs to _msg_ and should not be freed.
+
 `flux_event_unpack()` decodes a Flux event message with a JSON payload as
 above, parsing the payload using variable arguments with a format string
 in the style of jansson's `json_unpack()` (used internally). Decoding fails
@@ -48,6 +59,10 @@ if the message doesn't have a JSON payload.
 and optional JSON payload _json_str_.  The newly constructed message that
 is returned must be destroyed with `flux_msg_destroy()`.
 
+`flux_event_encode_raw()` encodes a Flux event message with topic
+string _topic_.  If _data_ is non-NULL, its contents will be used as
+the message payload, and the payload type set to raw.
+
 `flux_event_pack()` encodes a Flux event message with a JSON payload as
 above, encoding the payload using variable arguments with a format string
 in the style of jansson's `json_pack()` (used internally). Decoding fails

src/common/libflux/event.c

@@ -76,6 +76,35 @@ int flux_event_decode (const flux_msg_t *msg, const char **topic, const char **j
     return rc;
 }
 
+int flux_event_decode_raw (const flux_msg_t *msg, const char **topicp,
+                           const void **datap, int *lenp)
+{
+    const char *topic;
+    const void *data = NULL;
+    int len = 0;
+    int flags;
+    int rc = -1;
+
+    if (!datap || !lenp) {
+        errno = EINVAL;
+        goto done;
+    }
+    if (event_decode (msg, &topic) < 0)
+        goto done;
+    if (flux_msg_get_payload (msg, &flags, &data, &len) < 0) {
+        if (errno != EPROTO)
+            goto done;
+        errno = 0;
+    }
+    if (topicp)
+        *topicp = topic;
+    *datap = data;
+    *lenp = len;
+    rc = 0;
+done:
+    return rc;
+}
+
 static int flux_event_vunpack (const flux_msg_t *msg, const char **topic,
                                const char *fmt, va_list ap)
 {
@@ -138,6 +167,20 @@ flux_msg_t *flux_event_encode (const char *topic, const char *json_str)
     return NULL;
 }
 
+flux_msg_t *flux_event_encode_raw (const char *topic,
+                                   const void *data, int len)
+{
+    flux_msg_t *msg = flux_event_create (topic);
+    if (!msg)
+        goto error;
+    if (data && flux_msg_set_payload (msg, 0, data, len) < 0)
+        goto error;
+    return msg;
+error:
+    flux_msg_destroy (msg);
+    return NULL;
+}
+
 static flux_msg_t *flux_event_vpack (const char *topic,
                                      const char *fmt, va_list ap)
 {

src/common/libflux/event.h

@@ -34,6 +34,20 @@ flux_msg_t *flux_event_encode (const char *topic, const char *json_str);
  */
 flux_msg_t *flux_event_pack (const char *topic, const char *fmt, ...);
 
+/* Encode an event message with optional raw payload.
+ */
+flux_msg_t *flux_event_encode_raw (const char *topic,
+                                   const void *data, int len);
+
+/* Decode an event message, with optional raw payload.
+ * If topic is non-NULL, assign the event topic string.
+ * Data and len must be non-NULL and will be assigned the payload and length.
+ * If there is no payload, they will be assigned NULL and zero.
+ * Returns 0 on success, or -1 on failure with errno set.
+ */
+int flux_event_decode_raw (const flux_msg_t *msg, const char **topic,
+                           const void **data, int *len);
+
 #ifdef __cplusplus
 }
 #endif

src/common/libflux/test/event.c

@@ -8,6 +8,10 @@ int main (int argc, char *argv[])
     flux_msg_t *msg;
     const char *topic, *s;
     const char *json_str = "{\"a\":42}";
+    const char data[] = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+    int len = strlen (data);
+    const void *d;
+    int l;
     int i;
 
     plan (NO_PLAN);
@@ -16,6 +20,10 @@ int main (int argc, char *argv[])
     errno = 0;
     ok ((msg = flux_event_encode (NULL, json_str)) == NULL && errno == EINVAL,
         "flux_event_encode returns EINVAL with no topic string");
+    errno = 0;
+    ok ((msg = flux_event_encode_raw (NULL, data, len)) == NULL
+        && errno == EINVAL,
+        "flux_event_encode_raw topic=NULL fails with EINVAL");
 
     /* without payload */
     ok ((msg = flux_event_encode ("foo.bar", NULL)) != NULL,
@@ -55,6 +63,28 @@ int main (int argc, char *argv[])
         "unpacked payload matched packed");
     flux_msg_destroy (msg);
 
+    /* raw */
+    ok ((msg = flux_event_encode_raw ("foo.bar", data, len)) != NULL,
+        "flux_event_encode_raw works with payload");
+    d = NULL;
+    l = 0;
+    topic = NULL;
+    ok (flux_event_decode_raw (msg, &topic, &d, &l) == 0
+        && topic != NULL && strcmp (topic, "foo.bar") == 0
+        && d != NULL && len == len && memcmp (d, data, len) == 0,
+        "flux_event_decode_raw returns encoded topic and payload");
+    ok (flux_event_decode_raw (msg, NULL, &d, &l) == 0
+        && d != NULL && len == len && memcmp (d, data, len) == 0,
+        "flux_event_decode_raw topic=NULL returns encoded payload");
+
+    errno = 0;
+    ok (flux_event_decode_raw (msg, NULL, NULL, &l) < 0 && errno == EINVAL,
+        "flux_event_decode_raw data=NULL fails with EINVAL");
+    errno = 0;
+    ok (flux_event_decode_raw (msg, NULL, &d, NULL) < 0 && errno == EINVAL,
+        "flux_event_decode_raw len=NULL fails with EINVAL");
+    flux_msg_destroy (msg);
+
     done_testing();
     return (0);
 }