Begin publish/subscribe framework

Adding support for application capabilities
Discard incomplete messages after a while

Author: mic159

MeshBase.cpp

@@ -13,14 +13,16 @@
 #define TO_ADDRESS(x) (0xAA00000000LL + x)
 
 #define PEER_DISCOVERY_TIME 4000
-#define PEER_CHECK_TIME 4000
+#define CHECK_TIME 4000
 #define PEER_TIMEOUT 3
+#define ASSEMBLY_TIMEOUT 2
 
 MeshBase::MeshBase(uint8_t ce, uint8_t cs)
 : radio(ce, cs)
 , address(0)
 , last_broadcast_time(0)
-, last_peer_check_time(0)
+, last_check_time(0)
+, application_capabilities(0)
 {}
 
 void MeshBase::Begin()
@@ -56,23 +58,47 @@ void MeshBase::Update()
 		} while (!done);
 	}
 
-	// Update peers
-	if (millis() - last_peer_check_time > PEER_CHECK_TIME)
+	// Do periodic checks
+	if (millis() - last_check_time > CHECK_TIME)
 	{
-		LinkedList<Peer>::Node* current = peers.first;
-		while(current != NULL)
+		// Check for expired peers
 		{
-			current->item->time += 1;
-			if (current->item->time >= PEER_TIMEOUT)
+			LinkedList<Peer>::Node* current = peers.first;
+			while(current != NULL)
 			{
-				Serial.print("Lost Peer: ");
-				Serial.println(current->item->address, DEC);
-				current = peers.Remove(current);
-			} else {
-				current = current->next;
+				current->item->time += 1;
+				if (current->item->time >= PEER_TIMEOUT)
+				{
+					Serial.print("Lost Peer: ");
+					Serial.println(current->item->address, DEC);
+					current = peers.Remove(current);
+				} else {
+					current = current->next;
+				}
 			}
 		}
-		last_peer_check_time = millis();
+
+		// Check for expired packets
+		{
+			Message* current = assembly_list.first;
+			while(current != NULL)
+			{
+				current->age += 1;
+				if (current->age >= ASSEMBLY_TIMEOUT)
+				{
+					Serial.print("Dropped partial message. address=");
+					Serial.print(current->header.address_from, DEC);
+					Serial.print(" msg_id=");
+					Serial.print(current->header.msg_id);
+					Serial.print(" blocks_recieved=");
+					Serial.println(current->blocks_recieved);
+					current = assembly_list.Remove(current);
+				} else {
+					current = current->next;
+				}
+			}
+		}
+		last_check_time = millis();
 	}
 }
 
@@ -101,6 +127,7 @@ void MeshBase::Message::AddPart(const void* payload, uint8_t len, uint8_t part_n
 		header.split_more = false;
 		header.split_part = part_num;
 	}
+	age = 0;
 }
 
 bool MeshBase::Message::IsDone() const
@@ -110,16 +137,8 @@ bool MeshBase::Message::IsDone() const
 	// So if split_more is false, and we have the right number of blocks_recieved
 	// we are good to go.
 	if (!header.split_more && blocks_recieved > header.split_part) {
-		if (blocks_recieved > 1) {
-			Serial.print(" R IsDone() : id=");
-			Serial.print(header.msg_id);
-			Serial.println(" - True **");
-		}
 		return true;
 	}
-	Serial.print(" R IsDone() : id=");
-	Serial.print(header.msg_id);
-	Serial.println(" - False");
 	return false;
 }
 
@@ -175,11 +194,12 @@ void MeshBase::HandlePeerDiscovery(const MeshBase::MessageHeader* msg, const voi
 		Serial.print(" num_peers=");
 		Serial.println(pd->num_peers, DEC);
 		Peer* p = new Peer(msg->address_from);
+		p->Update(pd);
 		peers.Add(p);
 		OnNewPeer(p);
 	} else {
 		// Existing peer, reset timer
-		peer->time = 0;
+		peer->Update(pd);
 	}
 }
 
@@ -189,7 +209,7 @@ void MeshBase::SendPeerDiscovery()
 	MeshBase::PeerDiscoveryMessage payload;
 	payload.protocol_version = 1;
 	payload.network_capabilities = 0;
-	payload.application_capabilities = 0;
+	payload.application_capabilities = application_capabilities;
 	payload.num_peers = peers.length;
 	payload.uptime = millis() / 1000;
 	SendMessage(PEER_DISCOVERY, type_peer_discovery, &payload, sizeof(payload), true);
@@ -273,3 +293,9 @@ MeshBase::Peer* MeshBase::GetPeer(uint32_t a)
 	return NULL;
 }
 
+void MeshBase::Peer::Update(const PeerDiscoveryMessage* msg)
+{
+	application_capabilities = msg->application_capabilities;
+	time = 0;
+}
+

MeshBase.h

@@ -7,63 +7,82 @@
 
 #define PACKED __attribute__ ((packed))
 
+typedef uint32_t address_t;
+
 class MeshBase
 {
 public:
 	MeshBase(uint8_t ce, uint8_t cs);
 
-	struct Peer {
-		uint32_t	address;
-		uint16_t	time;
-		Peer(uint32_t address) : address(address), time(0) {}
-	};
-
 	struct MessageHeader {
 		uint8_t		protocol_version : 4;
 		uint8_t		ttl : 4;
 		uint8_t		msg_id;
 		bool		split_more : 1;
 		uint8_t		split_part : 7;
 		uint8_t		type;
-		uint32_t	address_from;
+		address_t	address_from;
 	} PACKED;
 
 	struct Message {
-		Message(const MessageHeader& a) : header(a), data(NULL), data_used(0), blocks_recieved(0), next(0), prev(0) {}
+		Message(const MessageHeader& a) : header(a), data(NULL), data_used(0), blocks_recieved(0), next(NULL), prev(NULL), age(0) {}
 		~Message();
 		MessageHeader header;
 		void* data;
 		uint8_t data_used;
 		uint8_t blocks_recieved;
 		Message* next;
 		Message* prev;
+		uint8_t age;
 
 		void AddPart(const void* data, uint8_t len, uint8_t part_num, bool more_parts);
 		bool IsDone() const;
 	};
 
 	// -- Message types --
-	enum message_type {
+	enum MessageType {
 		type_peer_discovery,
 		type_peer_list,
 		type_user,
 	};
 
+	enum ApplicationCapabilities {
+		capability_publish_events = 1 >> 0,
+	};
+
 	void			Begin();
 	void			Update();
-	void			SendMessage(uint32_t address, uint8_t type, const void* data, uint8_t length);
-	void			SendMessage(uint32_t address, uint8_t type, const void* data, uint8_t length, bool is_broadcast);
-	uint32_t		GetAddress() const { return address; }
+	void			SendMessage(address_t address, uint8_t type, const void* data, uint8_t length);
+	void			SendMessage(address_t address, uint8_t type, const void* data, uint8_t length, bool is_broadcast);
+	address_t		GetAddress() const { return address; }
 	bool			IsReady() const { return address != 0; }
 protected:
+	struct PeerDiscoveryMessage
+	{
+		uint8_t		protocol_version;
+		uint8_t		network_capabilities; // What routing/networking can I do for the network
+		uint8_t		application_capabilities; // What type of data do I expose
+		uint16_t	num_peers; // Number of direct peers
+		uint32_t	uptime; // Seconds since boot
+	} PACKED;
+
+	struct Peer {
+		Peer(uint32_t address) : address(address), time(0), application_capabilities(0) {}
+		uint32_t	address;
+		uint8_t		time;
+		uint8_t		application_capabilities;
+		void Update(const PeerDiscoveryMessage* msg);
+	};
+
 	virtual void	OnMessage(const MessageHeader* meta, const void* data, uint8_t length) = 0;
 	virtual void	OnNewPeer(Peer*) {}
 	virtual void	OnLostPeer(Peer*) {}
+	uint8_t			application_capabilities;
 private:
 	uint32_t		address;
 	RF24			radio;
 	unsigned long	last_broadcast_time;
-	unsigned long	last_peer_check_time;
+	unsigned long	last_check_time;
 
 	void			SendPeerDiscovery();
 	void			HandlePeerDiscovery(const MessageHeader* msg, const void* buff, uint8_t length);
@@ -75,15 +94,6 @@ class MeshBase
 
 	Peer*			GetPeer(uint32_t address);
 
-	struct PeerDiscoveryMessage
-	{
-		uint8_t		protocol_version;
-		uint8_t		network_capabilities; // What routing/networking can I do for the network
-		uint8_t		application_capabilities; // What type of data do I expose
-		uint16_t	num_peers; // Number of direct peers
-		uint32_t	uptime; // Seconds since boot
-	} PACKED;
-
 };
 
 #endif

Publisher.h

@@ -0,0 +1,48 @@
+#ifndef PUBLISHER_H
+#define PUBLISHER_H
+
+#include <Arduino.h>
+#include "MeshBase.h"
+
+class PublishApp : public MeshBase
+{
+public:
+	PublishApp() : MeshBase(9, 10)
+	{
+		application_capabilities |= MeshBase::capability_publish_events;
+	}
+
+	void OnEvent(uint8_t event_data)
+	{
+		const Target* current = targets.first;
+		while (current != NULL)
+		{
+			SendMessage(current->address, type_on_event, &event_data, sizeof(event_data));
+			current = current->next;
+		}
+	}
+
+	enum PublishMessageType {
+		type_on_event = MeshBase::type_user,
+		type_subscribe,
+	};
+protected:
+	virtual void	OnMessage(const MeshBase::MessageHeader* meta, const void* data, uint8_t length)
+	{
+		if (meta->type == type_subscribe)
+		{
+			targets.Add(new Target(meta->address_from));
+		}
+	}
+private:
+	struct Target
+	{
+		Target(address_t target) : address(target), prev(NULL), next(NULL) {}
+		address_t address;
+		Target* prev;
+		Target* next;
+	};
+	LinkedList2<Target> targets;
+};
+
+#endif // PUBLISHER_H

RF_test.ino

@@ -0,0 +1,33 @@
+#include <SPI.h>
+#include "RF24.h"
+#include "MeshBase.h"
+#include "Publisher.h"
+#include "LinkedList.h"
+
+PublishApp app;
+
+unsigned long last_time;
+uint8_t sequence;
+
+void setup()
+{
+    Serial.begin(19200);
+    Serial.println("Starting...");
+    randomSeed(analogRead(0));
+    app.Begin();
+    last_time = millis();
+    sequence = 0;
+}
+
+void loop()
+{
+    app.Update();
+    delay(100);
+    if (millis() - last_time > 10000)
+    {
+		app.OnEvent(sequence);
+		++sequence;
+		last_time = millis();
+    }
+}
+