diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2017-04-13 09:39:23 -0700 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2017-04-13 09:39:23 -0700 |
| commit | cb4ab189ce58333f6e5262a9972788864f003807 (patch) | |
| tree | 83eec38de00e0b291b87a52d7bc395e746ac0a4f /gr-zeromq/lib/base_impl.cc | |
| parent | c10208b11b4c5af9223f335bccb20bd809010ca8 (diff) | |
| parent | 3a2c67c67639e0a43ec90e69199c0e4b0cf1b05b (diff) | |
Merge branch 'master' into next
Diffstat (limited to 'gr-zeromq/lib/base_impl.cc')
| -rw-r--r-- | gr-zeromq/lib/base_impl.cc | 17 |
1 files changed, 15 insertions, 2 deletions
diff --git a/gr-zeromq/lib/base_impl.cc b/gr-zeromq/lib/base_impl.cc index f33315dd40..76baeafff2 100644 --- a/gr-zeromq/lib/base_impl.cc +++ b/gr-zeromq/lib/base_impl.cc @@ -165,6 +165,11 @@ namespace gr { if (!(items[0].revents & ZMQ_POLLIN)) return false; + /* Is this the start or continuation of a multi-part message? */ + int64_t more = 0; + size_t more_len = sizeof(more); + d_socket->getsockopt(ZMQ_RCVMORE, &more, &more_len); + /* Reset */ d_msg.rebuild(); d_tags.clear(); @@ -174,8 +179,8 @@ namespace gr { /* Get the message */ d_socket->recv(&d_msg); - /* Parse header */ - if (d_pass_tags) + /* Parse header from the first (or only) message of a multi-part message */ + if (d_pass_tags && !more) { uint64_t rcv_offset; @@ -188,6 +193,14 @@ namespace gr { } } + /* Each message must contain an integer mutliple of data vectors */ + if ((d_msg.size() - d_consumed_bytes) % d_vsize != 0) + { + throw std::runtime_error( + boost::str(boost::format("Incompatible vector sizes: " + "need a multiple of %1% bytes per message") % d_vsize)); + } + /* We got one ! */ return true; } |