1 files changed, 47 insertions, 13 deletions

diff --git a/src/munia_proto.cc b/src/munia_proto.cc
index 0508f9f..f0ec5be 100644
--- a/src/munia_proto.cc
+++ b/src/munia_proto.cc
@@ -98,8 +98,6 @@ int callback_lws_node(struct lws *wsi,
                       enum lws_callback_reasons reason,
                       void *user, void *in, size_t len)
 {
-	node_manager.tree.toStdOut();
-
 	DEBUG(proto, "Callback on %p\n", wsi);
 
 	switch (reason)
@@ -119,7 +117,8 @@ int callback_lws_node(struct lws *wsi,
 			// the position of the remove msg.
 			//
 			msgqueue_t::iterator i = msgqueue[wsi].begin();
-			while(i != msgqueue[wsi].end()) {
+			while(i != msgqueue[wsi].end())
+			{
 				message_t &msg = *i;
 				if(msg.cmd == cmd::remove)
 				{
@@ -168,10 +167,21 @@ int callback_lws_node(struct lws *wsi,
 					{
 						msgstr += " ";
 					}
-					auto msgs = msg_tostring(msg);
-					for(const auto& msg_string : msgs)
+					if(msg.cmd == cmd::error)
+					{
+						auto msgs = msg_tostring(msg);
+						for(const auto& msg_string : msgs)
+						{
+							msgstr += msg_string;
+						}
+					}
+					else
 					{
-						msgstr += std::to_string(msg.tid) + " " + msg_string;
+						auto msgs = msg_tostring(msg);
+						for(const auto& msg_string : msgs)
+						{
+							msgstr += std::to_string(msg.tid) + " " + msg_string;
+						}
 					}
 				}
 
@@ -216,7 +226,7 @@ int callback_lws_node(struct lws *wsi,
 			std::string data;
 			data.append((char*)in, len);
 
-			MessageList mlst = parse_msg(data);
+			MessageList mlst = parse_msg(data, wsi);
 			DEBUG(proto, "Handling %d incoming message\n", (int)mlst.size());
 			MessageList omsgs = handle_msg(mlst, wsi);
 			DEBUG(proto, "Handling %d outgoing messages\n", (int)omsgs.size());
@@ -224,21 +234,31 @@ int callback_lws_node(struct lws *wsi,
 			MessageList::iterator omi = omsgs.begin();
 			while(omi != omsgs.end())
 			{
-				DEBUG(proto, "Message\n");
+				DEBUG(proto, "Message %d\n", (int)omi->cmd);
 				if(omi->cmd == cmd::subscribe)
 				{
-					connection_handler.subscribe(wsi, omi->subscribe.id);
 					NodeIdList ids;
 					try
 					{
 						ids = node_manager.subNodes(omi->subscribe.id);
+						connection_handler.subscribe(wsi, omi->subscribe.id);
+					}
+					catch (const ErrorMessage& e)
+					{
+						// Set client id (wsi) and forward upstream
+						msgqueue[wsi].push_back(create_msg_error(e, wsi));
+						omi++;
+						continue;
 					}
-					catch(...)
+					catch (...)
 					{
-						DEBUG(proto, "No such node %d\n", (int)omi->subscribe.id);
+						msgqueue[wsi].push_back(
+							create_msg_error(ErrorCode::Unknown,
+							                 "Unknown subscribe error", wsi));
 						omi++;
 						continue;
 					}
+
 					NodeIdList::iterator id = ids.begin();
 					while(id != ids.end())
 					{
@@ -282,12 +302,22 @@ int callback_lws_node(struct lws *wsi,
 					{
 						ids = node_manager.subNodes(omi->unsubscribe.id);
 					}
-					catch(...)
+					catch (const ErrorMessage& e)
+					{
+						// Set client id (wsi) and forward upstream
+						msgqueue[wsi].push_back(create_msg_error(e, wsi));
+						omi++;
+						continue;
+					}
+					catch (...)
 					{
-						DEBUG(proto, "No such node %d\n", (int)omi->unsubscribe.id);
+						msgqueue[wsi].push_back(
+							create_msg_error(ErrorCode::Unknown,
+							                 "Unknown subscribe error", wsi));
 						omi++;
 						continue;
 					}
+
 					NodeIdList::reverse_iterator id = ids.rbegin();
 					while(id != ids.rend())
 					{
@@ -302,6 +332,10 @@ int callback_lws_node(struct lws *wsi,
 					}
 
 				}
+				else if(omi->cmd == cmd::error)
+				{
+					msgqueue[omi->error.wsi].push_back(*omi);
+				}
 				else
 				{
 					DEBUG(proto, "%d nodes affected by command\n",