3 files changed, 18 insertions, 7 deletions
diff --git a/gr-zeromq/lib/pull_msg_source_impl.cc b/gr-zeromq/lib/pull_msg_source_impl.cc
index 404333e891..651bc530fa 100644
--- a/gr-zeromq/lib/pull_msg_source_impl.cc
+++ b/gr-zeromq/lib/pull_msg_source_impl.cc
@@ -91,8 +91,12 @@ void pull_msg_source_impl::readloop()
 
             std::string buf(static_cast<char*>(msg.data()), msg.size());
             std::stringbuf sb(buf);
-            pmt::pmt_t m = pmt::deserialize(sb);
-            message_port_pub(d_port, m);
+            try {
+                pmt::pmt_t m = pmt::deserialize(sb);
+                message_port_pub(d_port, m);
+            } catch (pmt::exception& e) {
+                GR_LOG_ERROR(d_logger, std::string("Invalid PMT message: ") + e.what());
+            }
 
         } else {
             boost::this_thread::sleep(boost::posix_time::microseconds(100));
diff --git a/gr-zeromq/lib/req_msg_source_impl.cc b/gr-zeromq/lib/req_msg_source_impl.cc
index c246e3a57c..0118b4d084 100644
--- a/gr-zeromq/lib/req_msg_source_impl.cc
+++ b/gr-zeromq/lib/req_msg_source_impl.cc
@@ -109,8 +109,12 @@ void req_msg_source_impl::readloop()
 
             std::string buf(static_cast<char*>(msg.data()), msg.size());
             std::stringbuf sb(buf);
-            pmt::pmt_t m = pmt::deserialize(sb);
-            message_port_pub(d_port, m);
+            try {
+                pmt::pmt_t m = pmt::deserialize(sb);
+                message_port_pub(d_port, m);
+            } catch (pmt::exception& e) {
+                GR_LOG_ERROR(d_logger, std::string("Invalid PMT message: ") + e.what());
+            }
 
         } else {
             boost::this_thread::sleep(boost::posix_time::microseconds(100));
diff --git a/gr-zeromq/lib/sub_msg_source_impl.cc b/gr-zeromq/lib/sub_msg_source_impl.cc
index a8fd0a9117..3ead4025f1 100644
--- a/gr-zeromq/lib/sub_msg_source_impl.cc
+++ b/gr-zeromq/lib/sub_msg_source_impl.cc
@@ -89,9 +89,12 @@ void sub_msg_source_impl::readloop()
 #endif
             std::string buf(static_cast<char*>(msg.data()), msg.size());
             std::stringbuf sb(buf);
-            pmt::pmt_t m = pmt::deserialize(sb);
-
-            message_port_pub(d_port, m);
+            try {
+                pmt::pmt_t m = pmt::deserialize(sb);
+                message_port_pub(d_port, m);
+            } catch (pmt::exception& e) {
+                GR_LOG_ERROR(d_logger, std::string("Invalid PMT message: ") + e.what());
+            }
         } else {
             boost::this_thread::sleep(boost::posix_time::microseconds(100));
         }