Thread portable class Gate portable class Timer class Pool class Terimber 2.0 About C++

Downloads Products & Services Support Clients Open Source About Home / Open source / Terimber 2.0 msg_que.h Go to the documentation of this file. /* * The Software License * ================================================================================= * Copyright (c) 2003-.The Terimber Corporation. All rights reserved. * ================================================================================= * Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * The end-user documentation included with the redistribution, if any, * must include the following acknowledgment: * "This product includes software developed by the Terimber Corporation." * ================================================================================= * THIS SOFTWARE IS PROVIDED "AS IS" AND ANY EXPRESSED OR IMPLIED WARRANTIES, * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. * IN NO EVENT SHALL THE TERIMBER CORPORATION OR ITS CONTRIBUTORS BE LIABLE FOR * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * ================================================================================ */ #ifndef _terimber_msg_que_h_ #define _terimber_msg_que_h_ #include "base/except.h" #include "base/primitives.h" #include "threadpool/threadpoolfactory.h" #include "aiomsg/msg_cpp.h" #include "aiomsg/msg_base.h" BEGIN_TERIMBER_NAMESPACE #pragma pack(4) template < size_t P = 3, size_t C = 1024 > class msg_queue { enum en_msg_queue { PRIORITY = P, CAPACITY = C }; public: inline bool block(); inline bool is_block(); protected: msg_queue(); inline bool pop( msg_cpp*& item ); inline bool peek(); inline bool touch( size_t& top_priority ); virtual void wakeup() = 0; inline void push( msg_cpp* item ); inline bool unblock(); private: static inline ub1_t _check( msg_cpp* item ); private: bool _blocked; mutex _mtx_queue; list< msg_cpp* > _queue[PRIORITY]; }; template < size_t P, size_t C > msg_queue< P, C >::msg_queue() : _blocked(false) { } // static template < size_t P, size_t C > inline ub1_t msg_queue< P, C >::_check(msg_cpp* item) { return item->priority >= PRIORITY ? PRIORITY - 1 : (ub1_t)item->priority; } template < size_t P, size_t C > inline void msg_queue< P, C >::push(msg_cpp* item) { // locks mutex mutex_keeper keeper(_mtx_queue); if (_blocked) // queue blocked exception::_throw("Queue has been blocked"); // gets the correspondent queue list< msg_cpp* >& q = _queue[_check(item)]; if (q.size() == CAPACITY) // out of space exception::_throw("Queue max capacity has been reached"); // adds item to queue q.push_back(item); // wakes up thread wakeup(); } template < size_t P, size_t C > inline bool msg_queue< P, C >::pop(msg_cpp*& item) { // locks mutex mutex_keeper keeper(_mtx_queue); // loop for all queues for (ub1_t index = 0; index < PRIORITY; ++index) { if (!_queue[index].empty()) { // gets the message pointer item = _queue[index].front(); // removes message from queue _queue[index].pop_front(); return true; } } return false; } template < size_t P, size_t C > inline bool msg_queue< P, C >::peek() { // locks mutex mutex_keeper keeper(_mtx_queue); // loop for all queues for (ub1_t index = 0; index < PRIORITY; ++index) { if (!_queue[index].empty()) return true; } return false; } template < size_t P, size_t C > inline bool msg_queue< P, C >::touch(size_t& top_priority) { // locks mutex mutex_keeper keeper(_mtx_queue); // loop for all queues for (ub1_t index = 0; index < PRIORITY; ++index) { if (!_queue[index].empty()) { top_priority = index; return true; } } return false; } template < size_t P, size_t C > inline bool msg_queue< P, C >::block() { // locks mutex mutex_keeper keeper(_mtx_queue); return !_blocked ? (_blocked = true) : false; } template < size_t P, size_t C > inline bool msg_queue< P, C >::unblock() { // locks mutex mutex_keeper keeper(_mtx_queue); return _blocked ? !(_blocked = false) : false; } template < size_t P, size_t C > inline bool msg_queue< P, C >::is_block() { // locks mutex mutex_keeper keeper(_mtx_queue); return _blocked; } // forwards declaration class msg_communicator; // and msg_queue class class msg_queue_processor : public msg_base, public msg_queue< 3 >, public terimber_thread_employer { public: msg_queue_processor(msg_communicator* communicator ); ~msg_queue_processor(); protected: virtual bool v_has_job( size_t ident, void* user_data ); protected: virtual void v_on(); virtual void v_off(); }; #pragma pack() END_TERIMBER_NAMESPACE #endif // _terimber_msg_que_h_

© Copyright Terimber 2003-.