villas-binary: make source_index configurable

include/villas/formats/msg.hpp

@@ -50,7 +50,7 @@ void msg_hdr_ntoh(struct msg *m);
 int msg_verify(const struct msg *m);
 
 /** Copy fields from \p msg into \p smp. */
-int msg_to_sample(const struct msg *msg, struct sample *smp, const struct vlist *sigs);
+int msg_to_sample(const struct msg *msg, struct sample *smp, const struct vlist *sigs, uint8_t *source_index);
 
 /** Copy fields form \p smp into \p msg. */
-int msg_from_sample(struct msg *msg, const struct sample *smp, const struct vlist *sigs);
+int msg_from_sample(struct msg *msg, const struct sample *smp, const struct vlist *sigs, uint8_t source_index);

include/villas/formats/msg_format.hpp

@@ -65,7 +65,7 @@ struct msg
   #error Invalid byte-order
 #endif
 
-	uint8_t id;			/**< An id which identifies the source of this sample. */
+	uint8_t source_index;	/**< An id which identifies the source of this sample. */
 	uint16_t length;	/**< The number of values in msg::data[]. */
 	uint32_t sequence;	/**< The sequence number is incremented by one for consecutive messages. */
 

include/villas/formats/villas_binary.hpp

@@ -36,16 +36,24 @@ namespace node {
 
 class VillasBinaryFormat : public BinaryFormat {
 
-public:
+protected:
+	uint8_t source_index;
 	bool web;
 
-	VillasBinaryFormat(int fl, bool w) :
+public:
+	VillasBinaryFormat(int fl, bool w, uint8_t sid = 0) :
 		BinaryFormat(fl),
+		source_index(sid),
 		web(w)
 	{ }
 
+	virtual
 	int sscan(const char *buf, size_t len, size_t *rbytes, struct sample * const smps[], unsigned cnt);
+	virtual
 	int sprint(char *buf, size_t len, size_t *wbytes, const struct sample * const smps[], unsigned cnt);
+
+	virtual
+	void parse(json_t *json);
 };
 
 
@@ -55,14 +63,15 @@ class VillasBinaryFormatPlugin : public FormatFactory {
 public:
 	using FormatFactory::FormatFactory;
 
-	virtual Format * make()
+	virtual
+	Format * make()
 	{
 		return new VillasBinaryFormat((int) SampleFlags::HAS_TS_ORIGIN | (int) SampleFlags::HAS_SEQUENCE | (int) SampleFlags::HAS_DATA, web);
 	}
 
 	/// Get plugin name
-	virtual std::string
-	getName() const
+	virtual
+	std::string getName() const
 	{
 		std::stringstream ss;
 
@@ -72,8 +81,8 @@ public:
 	}
 
 	/// Get plugin description
-	virtual std::string
-	getDescription() const
+	virtual
+	std::string getDescription() const
 	{
 		std::stringstream ss;
 

lib/formats/msg.cpp

@@ -73,7 +73,7 @@ int msg_verify(const struct msg *m)
 		return 0;
 }
 
-int msg_to_sample(const struct msg *msg, struct sample *smp, const struct vlist *sigs)
+int msg_to_sample(const struct msg *msg, struct sample *smp, const struct vlist *sigs, uint8_t *source_index)
 {
 	int ret;
 	unsigned i;
@@ -107,15 +107,18 @@ int msg_to_sample(const struct msg *msg, struct sample *smp, const struct vlist
 	smp->sequence = msg->sequence;
 	MSG_TS(msg, smp->ts.origin);
 
+	if (source_index)
+		*source_index = msg->source_index;
+
 	return 0;
 }
 
-int msg_from_sample(struct msg *msg_in, const struct sample *smp, const struct vlist *sigs)
+int msg_from_sample(struct msg *msg_in, const struct sample *smp, const struct vlist *sigs, uint8_t source_index)
 {
 	msg_in->type     = MSG_TYPE_DATA;
 	msg_in->version  = MSG_VERSION;
 	msg_in->reserved1 = 0;
-	msg_in->id = 0;
+	msg_in->source_index = source_index;
 	msg_in->length   = (uint16_t) smp->length;
 	msg_in->sequence = (uint32_t) smp->sequence;
 	msg_in->ts.sec  = smp->ts.origin.tv_sec;

lib/formats/villas_binary.cpp

@@ -27,8 +27,10 @@
 #include <villas/formats/msg.hpp>
 #include <villas/formats/msg_format.hpp>
 #include <villas/sample.h>
+#include <villas/exceptions.hpp>
 #include <villas/utils.hpp>
 
+using namespace villas;
 using namespace villas::node;
 
 int VillasBinaryFormat::sprint(char *buf, size_t len, size_t *wbytes, const struct sample * const smps[], unsigned cnt)
@@ -44,7 +46,7 @@ int VillasBinaryFormat::sprint(char *buf, size_t len, size_t *wbytes, const stru
 		if (ptr + MSG_LEN(smp->length) > buf + len)
 			break;
 
-		ret = msg_from_sample(msg, smp, smp->signals);
+		ret = msg_from_sample(msg, smp, smp->signals, source_index);
 		if (ret)
 			return ret;
 
@@ -68,6 +70,7 @@ int VillasBinaryFormat::sscan(const char *buf, size_t len, size_t *rbytes, struc
 	int ret, values;
 	unsigned i = 0;
 	const char *ptr = buf;
+	uint8_t sid; // source_index
 
 	if (len % 4 != 0)
 		return -1; /* Packet size is invalid: Must be multiple of 4 bytes */
@@ -90,7 +93,7 @@ int VillasBinaryFormat::sscan(const char *buf, size_t len, size_t *rbytes, struc
 
 		/* Check if remainder of message is in buffer boundaries */
 		if (ptr + MSG_LEN(values) > buf + len)
-			return -3; /*Invalid msg receive */
+			return -3; /* Invalid msg receive */
 
 		if (web) {
 			/** @todo convert from little endian */
@@ -98,10 +101,13 @@ int VillasBinaryFormat::sscan(const char *buf, size_t len, size_t *rbytes, struc
 		else
 			msg_ntoh(msg);
 
-		ret = msg_to_sample(msg, smp, signals);
+		ret = msg_to_sample(msg, smp, signals, &sid);
 		if (ret)
 			return ret; /* Invalid msg received */
 
+		if (sid != source_index)
+			return -6; // source_index mismatch
+
 		ptr += MSG_LEN(smp->length);
 	}
 
@@ -111,5 +117,23 @@ int VillasBinaryFormat::sscan(const char *buf, size_t len, size_t *rbytes, struc
 	return i;
 }
 
+void VillasBinaryFormat::parse(json_t *json)
+{
+	int ret;
+	json_error_t err;
+	int sid = -1;
+
+	ret = json_unpack_ex(json, &err, 0, "{ s?: i }",
+		"source_index", &sid
+	);
+	if (ret)
+		throw ConfigError(json, err, "node-config-format-villas-binary", "Failed to parse format configuration");
+
+	if (sid >= 0)
+		source_index = sid;
+
+	Format::parse(json);
+}
+
 static VillasBinaryFormatPlugin<false> p1;
 static VillasBinaryFormatPlugin<true> p2;