diff options
| author | Johnathan Corgan <johnathan@corganlabs.com> | 2013-03-15 09:57:31 -0700 |
|---|---|---|
| committer | Johnathan Corgan <johnathan@corganlabs.com> | 2013-03-15 09:57:31 -0700 |
| commit | e965a5bb209ad46a509ccd21f393667fd69d95f9 (patch) | |
| tree | 40d683ef8f80980f12eac6cfe7f3423c49b45f87 /gr-blocks/lib/message_sink_impl.cc | |
| parent | 2bf9c4cb4b0b426690f353fc1662a13e70c0d5e0 (diff) | |
| parent | 27990ca9e236931e39a830e48f0a1efe13ec085f (diff) | |
Merge branch 'ofdm-master' into ofdm-next
Added fixups for next branch changes
Conflicts:
CMakeLists.txt
gnuradio-core/src/lib/io/gr_message_sink.cc
gnuradio-core/src/lib/io/gr_message_sink.h
gnuradio-core/src/lib/io/gr_message_sink.i
gnuradio-core/src/lib/io/gr_message_source.cc
gnuradio-core/src/lib/io/gr_message_source.h
gnuradio-core/src/lib/io/gr_message_source.i
gr-blocks/CMakeLists.txt
gr-digital/CMakeLists.txt
gr-digital/grc/digital_block_tree.xml
gr-digital/include/digital/CMakeLists.txt
gr-digital/include/digital_ofdm_cyclic_prefixer.h
gr-digital/lib/CMakeLists.txt
gr-digital/lib/digital_ofdm_cyclic_prefixer.cc
gr-digital/lib/ofdm_cyclic_prefixer_impl.h
gr-digital/python/CMakeLists.txt
gr-digital/swig/CMakeLists.txt
gr-digital/swig/digital_swig.i
Diffstat (limited to 'gr-blocks/lib/message_sink_impl.cc')
| -rw-r--r-- | gr-blocks/lib/message_sink_impl.cc | 79 |
1 files changed, 64 insertions, 15 deletions
diff --git a/gr-blocks/lib/message_sink_impl.cc b/gr-blocks/lib/message_sink_impl.cc index a8dbfb4c71..fbc7b27d58 100644 --- a/gr-blocks/lib/message_sink_impl.cc +++ b/gr-blocks/lib/message_sink_impl.cc @@ -44,11 +44,30 @@ namespace gr { (new message_sink_impl(itemsize, msgq, dont_block)); } + message_sink::sptr + message_sink::make(size_t itemsize, gr_msg_queue_sptr msgq, bool dont_block, + const std::string& lengthtagname) + { + return gnuradio::get_initial_sptr + (new message_sink_impl(itemsize, msgq, dont_block, lengthtagname)); + } + message_sink_impl::message_sink_impl(size_t itemsize, gr_msg_queue_sptr msgq, bool dont_block) : gr_sync_block("message_sink", gr_make_io_signature(1, 1, itemsize), gr_make_io_signature(0, 0, 0)), - d_itemsize(itemsize), d_msgq(msgq), d_dont_block(dont_block) + d_itemsize(itemsize), d_msgq(msgq), d_dont_block(dont_block), + d_tags(false), d_items_read(0) + { + } + + message_sink_impl::message_sink_impl(size_t itemsize, gr_msg_queue_sptr msgq, bool dont_block, + const std::string& lengthtagname) + : gr_sync_block("message_sink", + gr_make_io_signature(1, 1, itemsize), + gr_make_io_signature(0, 0, 0)), + d_itemsize(itemsize), d_msgq(msgq), d_dont_block(dont_block), + d_tags(true), d_lengthtagname(lengthtagname), d_items_read(0) { } @@ -61,23 +80,53 @@ namespace gr { gr_vector_const_void_star &input_items, gr_vector_void_star &output_items) { - const char *in = (const char*)input_items[0]; - - // if we'd block, drop the data on the floor and say everything is OK - if(d_dont_block && d_msgq->full_p()) - return noutput_items; - - // build a message to hold whatever we've got - gr_message_sptr msg = gr_make_message(0, // msg type - d_itemsize, // arg1 for other end - noutput_items, // arg2 for other end (redundant) - noutput_items * d_itemsize); // len of msg - memcpy(msg->msg(), in, noutput_items * d_itemsize); + const char *in = (const char *) input_items[0]; - d_msgq->handle(msg); // send it + if (d_tags) { + long packet_length = 0; + std::vector<gr_tag_t> tags; + this->get_tags_in_range(tags, 0, d_items_read, d_items_read+1); + //const size_t ninput_items = noutput_items; //assumption for sync block, this can change + for (unsigned int i = 0; i < tags.size(); i++) { + if (pmt::symbol_to_string(tags[i].key) == d_lengthtagname) { + packet_length = pmt::to_long(tags[i].value); + } + } + assert(packet_length != 0); + + // FIXME run this multiple times if input_items >= N * packet_length + if (noutput_items >= packet_length ) { + // If the message queue is full we drop the packet. + if (!d_msgq->full_p()) { + gr_message_sptr msg = gr_make_message(0, // msg type + d_itemsize, // arg1 for other end + packet_length, // arg2 for other end (redundant) + packet_length * d_itemsize); // len of msg + memcpy(msg->msg(), in, packet_length * d_itemsize); + d_msgq->handle(msg); // send it + } + d_items_read += packet_length; + return packet_length; + } else { + return 0; + } + } else { + // If the queue if full we drop all the data we got. + if (!d_msgq->full_p()) { + // build a message to hold whatever we've got + gr_message_sptr msg = gr_make_message(0, // msg type + d_itemsize, // arg1 for other end + noutput_items, // arg2 for other end (redundant) + noutput_items * d_itemsize); // len of msg + memcpy(msg->msg(), in, noutput_items * d_itemsize); + + d_msgq->handle(msg); // send it + } - return noutput_items; + return noutput_items; + } } } /* namespace blocks */ } /* namespace gr */ + |