Enable MSGQ

messaging/__init__.py

@@ -158,7 +158,7 @@ def __getitem__(self, s):
   def update(self, timeout=1000):
     msgs = []
     for sock in self.poller.poll(timeout):
-      msgs.append(recv_one(sock))
+      msgs.append(recv_one_or_none(sock))
     self.update_msgs(sec_since_boot(), msgs)
 
   def update_msgs(self, cur_time, msgs):

messaging/messaging.cc

@@ -1,13 +1,24 @@
 #include "messaging.hpp"
 #include "impl_zmq.hpp"
+#include "impl_msgq.hpp"
 
 Context * Context::create(){
-  Context * c = new ZMQContext();
+  Context * c;
+  if (std::getenv("ZMQ")){
+    c = new ZMQContext();
+  } else {
+    c = new MSGQContext();
+  }
   return c;
 }
 
 SubSocket * SubSocket::create(){
-  SubSocket * s = new ZMQSubSocket();
+  SubSocket * s;
+  if (std::getenv("ZMQ")){
+    s = new ZMQSubSocket();
+  } else {
+    s = new MSGQSubSocket();
+  }
   return s;
 }
 
@@ -26,7 +37,12 @@ SubSocket * SubSocket::create(Context * context, std::string endpoint, std::stri
 }
 
 PubSocket * PubSocket::create(){
-  PubSocket * s = new ZMQPubSocket();
+  PubSocket * s;
+  if (std::getenv("ZMQ")){
+    s = new ZMQPubSocket();
+  } else {
+    s = new MSGQPubSocket();
+  }
   return s;
 }
 
@@ -37,7 +53,12 @@ PubSocket * PubSocket::create(Context * context, std::string endpoint){
 }
 
 Poller * Poller::create(){
-  Poller * p = new ZMQPoller();
+  Poller * p;
+  if (std::getenv("ZMQ")){
+    p = new ZMQPoller();
+  } else {
+    p = new MSGQPoller();
+  }
   return p;
 }