![]() |
libzmq master
The Intelligent Transport Layer
|
00001 /* 00002 Copyright (c) 2010-2011 250bpm s.r.o. 00003 Copyright (c) 2010-2011 Other contributors as noted in the AUTHORS file 00004 00005 This file is part of 0MQ. 00006 00007 0MQ is free software; you can redistribute it and/or modify it under 00008 the terms of the GNU Lesser General Public License as published by 00009 the Free Software Foundation; either version 3 of the License, or 00010 (at your option) any later version. 00011 00012 0MQ is distributed in the hope that it will be useful, 00013 but WITHOUT ANY WARRANTY; without even the implied warranty of 00014 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 00015 GNU Lesser General Public License for more details. 00016 00017 You should have received a copy of the GNU Lesser General Public License 00018 along with this program. If not, see <http://www.gnu.org/licenses/>. 00019 */ 00020 00021 #ifndef __ZMQ_XSUB_HPP_INCLUDED__ 00022 #define __ZMQ_XSUB_HPP_INCLUDED__ 00023 00024 #include "socket_base.hpp" 00025 #include "session_base.hpp" 00026 #include "dist.hpp" 00027 #include "fq.hpp" 00028 #include "trie.hpp" 00029 #include "msg.hpp" 00030 00031 namespace zmq 00032 { 00033 00034 class xsub_t : 00035 public socket_base_t 00036 { 00037 public: 00038 00039 xsub_t (class ctx_t *parent_, uint32_t tid_); 00040 ~xsub_t (); 00041 00042 protected: 00043 00044 // Overloads of functions from socket_base_t. 00045 void xattach_pipe (class pipe_t *pipe_); 00046 int xsend (class msg_t *msg_, int flags_); 00047 bool xhas_out (); 00048 int xrecv (class msg_t *msg_, int flags_); 00049 bool xhas_in (); 00050 void xread_activated (class pipe_t *pipe_); 00051 void xwrite_activated (class pipe_t *pipe_); 00052 void xhiccuped (pipe_t *pipe_); 00053 void xterminated (class pipe_t *pipe_); 00054 00055 private: 00056 00057 // Check whether the message matches at least one subscription. 00058 bool match (class msg_t *msg_); 00059 00060 // Function to be applied to the trie to send all the subsciptions 00061 // upstream. 00062 static void send_subscription (unsigned char *data_, size_t size_, 00063 void *arg_); 00064 00065 // Fair queueing object for inbound pipes. 00066 fq_t fq; 00067 00068 // Object for distributing the subscriptions upstream. 00069 dist_t dist; 00070 00071 // The repository of subscriptions. 00072 trie_t subscriptions; 00073 00074 // If true, 'message' contains a matching message to return on the 00075 // next recv call. 00076 bool has_message; 00077 msg_t message; 00078 00079 // If true, part of a multipart message was already received, but 00080 // there are following parts still waiting. 00081 bool more; 00082 00083 xsub_t (const xsub_t&); 00084 const xsub_t &operator = (const xsub_t&); 00085 }; 00086 00087 class xsub_session_t : public session_base_t 00088 { 00089 public: 00090 00091 xsub_session_t (class io_thread_t *io_thread_, bool connect_, 00092 class socket_base_t *socket_, const options_t &options_, 00093 const char *protocol_, const char *address_); 00094 ~xsub_session_t (); 00095 00096 private: 00097 00098 xsub_session_t (const xsub_session_t&); 00099 const xsub_session_t &operator = (const xsub_session_t&); 00100 }; 00101 00102 } 00103 00104 #endif 00105