samples: net/nats: Fix parsing of MSG messages

This addresses the issues found by QA in ZEP-1012 and clarify the
documentated behavior as described in ZEP-1859.

Change-Id: I602e5749db7f6f44cf5be449b8e6f0d2ba66b69b
Signed-off-by: Leandro Pereira <[email protected]>

Author: lpereira

samples/net/nats/README.rst

@@ -131,15 +131,19 @@ If ``queue_group`` is NULL, it's not sent to the server.
 
 ``max_msgs`` specifies the number of messages that the server will
 send before actually unsubscribing the message. Can be 0 to
-immediately unsubscribe.
+immediately unsubscribe.  (See note below.)
 
 Both of these functions will return ``-ENOMEM`` if they couldn't build
 the message to transmit to the server. They can also return any error
 that ``net_context_send()`` can return.
 
-In order to conserve resources, the Zephyr implementation will not make
-not of subscribed topics. This is left as a task for the user of the API
-to handle, for instance, when the ``on_message()`` callback is called.
+Note:  In order to conserve resources, the library implementation will not
+make note of subscribed topics.  Both ``nats_subscribe()`` and
+``nats_unsubscribe()`` functions will only notify the server that the client
+is either interested or uninterested in a particular topic.  The
+``on_message()`` callback may be called to notify of changes in topics that
+have not been subscribed to (or have been recently unsubscribed).  It's up
+to the application to decide to ignore the message.
 
 Topics can be published by using the following function:
 
@@ -155,5 +159,5 @@ which case, subscribers to this topic won't receive this information as
 well.
 
 As ``net_subscribe()`` and ``net_unsubscribe()``, this function can
-return ``ENOMEM`` -or any other errors that ``net_context_send()``
+return ``-ENOMEM`` or any other errors that ``net_context_send()``
 returns.

samples/net/nats/src/main.c

@@ -203,6 +203,8 @@ static void write_led(const struct nats *nats,
 	pubstate = state ? "on" : "off";
 	nats_publish(nats, "led0", 0, msg->reply_to, 0,
 		     pubstate, strlen(pubstate));
+
+	printk("*** Turning LED %s\n", pubstate);
 }
 
 static int on_msg_received(const struct nats *nats,

samples/net/nats/src/nats.c

@@ -6,7 +6,9 @@
 
 #include <ctype.h>
 #include <errno.h>
+#include <json.h>
 #include <misc/printk.h>
+#include <misc/util.h>
 #include <net/nbuf.h>
 #include <net/net_context.h>
 #include <net/net_core.h>
@@ -17,10 +19,11 @@
 #include <stdlib.h>
 #include <string.h>
 #include <zephyr.h>
-#include <json.h>
 
 #include "nats.h"
 
+#define CMD_BUF_LEN 256
+
 struct nats_info {
 	const char *server_id;
 	const char *version;
@@ -128,7 +131,8 @@ static inline int transmit(struct net_context *conn, const char buffer[],
 	.offset = offsetof(struct_, member_), \
 	.type = type_ \
 }
-static int handle_server_info(struct nats *nats, char *payload, size_t len)
+static int handle_server_info(struct nats *nats, char *payload, size_t len,
+			      struct net_buf *buf, uint16_t offset)
 {
 	static const struct json_obj_descr descr[] = {
 		FIELD(struct nats_info, server_id, JSON_TOK_STRING),
@@ -227,10 +231,42 @@ static char *strsep(char *strp, const char *delims)
 	return NULL;
 }
 
-static int handle_server_msg(struct nats *nats, char *payload, size_t len)
+static int copy_nbuf_to_buf(struct net_buf *src, uint16_t offset,
+			    char *dst, size_t dst_size, size_t n_bytes)
+{
+	uint16_t to_copy;
+	uint16_t copied;
+
+	if (dst_size < n_bytes) {
+		return -ENOMEM;
+	}
+
+	while (src && offset >= src->len) {
+		offset -= src->len;
+		src = src->frags;
+	}
+
+	for (copied = 0; src && n_bytes > 0; offset = 0) {
+		to_copy = min(n_bytes, src->len - offset);
+
+		memcpy(dst + copied, (char *)src->data + offset, to_copy);
+		copied += to_copy;
+
+		n_bytes -= to_copy;
+		src = src->frags;
+	}
+
+	if (n_bytes > 0) {
+		return -ENOMEM;
+	}
+
+	return 0;
+}
+
+static int handle_server_msg(struct nats *nats, char *payload, size_t len,
+			     struct net_buf *buf, uint16_t offset)
 {
-	char *subject, *sid, *reply_to, *bytes;
-	char *end_ptr;
+	char *subject, *sid, *reply_to, *bytes, *end_ptr;
 	char prev_end = payload[len];
 	size_t payload_size;
 
@@ -241,21 +277,21 @@ static int handle_server_msg(struct nats *nats, char *payload, size_t len)
 	subject = payload;
 	sid = strsep(subject, " \t");
 	reply_to = strsep(sid, " \t");
-	if (!reply_to) {
-		bytes = strsep(sid, "\r");
-	} else {
-		bytes = strsep(reply_to, "\r");
-	}
-
-	payload[len] = prev_end;
+	bytes = strsep(reply_to, " \t");
 
 	if (!bytes) {
-		return -EINVAL;
+		if (!reply_to) {
+			return -EINVAL;
+		}
+
+		bytes = reply_to;
+		reply_to = NULL;
 	}
 
 	/* Parse the payload size */
 	errno = 0;
 	payload_size = strtoul(bytes, &end_ptr, 10);
+	payload[len] = prev_end;
 	if (errno != 0) {
 		return -errno;
 	}
@@ -264,29 +300,35 @@ static int handle_server_msg(struct nats *nats, char *payload, size_t len)
 		return -EINVAL;
 	}
 
-	if (payload_size >= payload + len - end_ptr) {
-		return -EINVAL;
+	if (payload_size >= CMD_BUF_LEN - len) {
+		return -ENOMEM;
 	}
 
-	payload = end_ptr + 2;
+	if (copy_nbuf_to_buf(buf, offset, end_ptr, CMD_BUF_LEN - len,
+			     payload_size) < 0) {
+		return -ENOMEM;
+	}
+	end_ptr[payload_size] = '\0';
 
 	return nats->on_message(nats, &(struct nats_msg) {
 		.subject = subject,
 		.sid = sid,
 		.reply_to = reply_to,
-		.payload = payload,
+		.payload = end_ptr,
 		.payload_len = payload_size,
 	});
 }
 
-static int handle_server_ping(struct nats *nats, char *payload, size_t len)
+static int handle_server_ping(struct nats *nats, char *payload, size_t len,
+			      struct net_buf *buf, uint16_t offset)
 {
 	static const char pong[] = "PONG\r\n";
 
 	return transmit(nats->conn, pong, sizeof(pong) - 1);
 }
 
-static int ignore(struct nats *nats, char *payload, size_t len)
+static int ignore(struct nats *nats, char *payload, size_t len,
+		  struct net_buf *buf, uint16_t offset)
 {
 	/* FIXME: Notify user of success/errors.  This would require
 	 * maintaining information of what was the last sent command in
@@ -302,12 +344,14 @@ static int ignore(struct nats *nats, char *payload, size_t len)
 	.len = sizeof(cmd_) - 1, \
 	.handle = handler_ \
 }
-static int handle_server_cmd(struct nats *nats, char *cmd, size_t len)
+static int handle_server_cmd(struct nats *nats, char *cmd, size_t len,
+			     struct net_buf *buf, uint16_t offset)
 {
 	static const struct {
 		const char *op;
 		size_t len;
-		int (*handle)(struct nats *nats, char *payload, size_t len);
+		int (*handle)(struct nats *nats, char *payload, size_t len,
+			      struct net_buf *buf, uint16_t offset);
 	} cmds[] = {
 		CMD("INFO", handle_server_info),
 		CMD("MSG", handle_server_msg),
@@ -327,15 +371,16 @@ static int handle_server_cmd(struct nats *nats, char *cmd, size_t len)
 		}
 	}
 	payload_len = len - (size_t)(payload - cmd);
-	len = (size_t)(payload - cmd);
+	len = (size_t)(payload - cmd - 1);
 
 	for (i = 0; i < ARRAY_SIZE(cmds); i++) {
 		if (len != cmds[i].len) {
 			continue;
 		}
 
 		if (!strncmp(cmds[i].op, cmd, len)) {
-			return cmds[i].handle(nats, payload, payload_len);
+			return cmds[i].handle(nats, payload, payload_len,
+					      buf, offset);
 		}
 	}
 
@@ -490,7 +535,7 @@ static void receive_cb(struct net_context *ctx, struct net_buf *buf, int status,
 		       void *user_data)
 {
 	struct nats *nats = user_data;
-	char cmd_buf[256];
+	char cmd_buf[CMD_BUF_LEN];
 	struct net_buf *tmp;
 	uint16_t pos = 0, cmd_len = 0;
 	size_t len;
@@ -513,7 +558,7 @@ static void receive_cb(struct net_context *ctx, struct net_buf *buf, int status,
 	while (tmp) {
 		len = tmp->len - pos;
 
-		end_of_line = memchr((uint8_t *)tmp->data + pos, '\r', len);
+		end_of_line = memchr((uint8_t *)tmp->data + pos, '\n', len);
 		if (end_of_line) {
 			len = end_of_line - ((uint8_t *)tmp->data + pos);
 		}
@@ -526,12 +571,17 @@ static void receive_cb(struct net_context *ctx, struct net_buf *buf, int status,
 		cmd_len += len;
 
 		if (end_of_line) {
+			int ret;
+
 			if (tmp) {
 				tmp = net_nbuf_read(tmp, pos, &pos, 1, NULL);
 			}
 
 			cmd_buf[cmd_len] = '\0';
-			if (handle_server_cmd(nats, cmd_buf, cmd_len) < 0) {
+
+			ret = handle_server_cmd(nats, cmd_buf, cmd_len,
+						tmp, pos);
+			if (ret < 0) {
 				/* FIXME: What to do with unhandled messages? */
 				break;
 			}