![]() |
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_XPUB_HPP_INCLUDED__ 00022 #define __ZMQ_XPUB_HPP_INCLUDED__ 00023 00024 #include <deque> 00025 #include <string> 00026 00027 #include "socket_base.hpp" 00028 #include "session_base.hpp" 00029 #include "mtrie.hpp" 00030 #include "array.hpp" 00031 #include "dist.hpp" 00032 00033 namespace zmq 00034 { 00035 00036 class xpub_t : 00037 public socket_base_t 00038 { 00039 public: 00040 00041 xpub_t (class ctx_t *parent_, uint32_t tid_); 00042 ~xpub_t (); 00043 00044 // Implementations of virtual 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 xterminated (class pipe_t *pipe_); 00053 00054 private: 00055 00056 // Function to be applied to the trie to send all the subsciptions 00057 // upstream. 00058 static void send_unsubscription (unsigned char *data_, size_t size_, 00059 void *arg_); 00060 00061 // Function to be applied to each matching pipes. 00062 static void mark_as_matching (class pipe_t *pipe_, void *arg_); 00063 00064 // List of all subscriptions mapped to corresponding pipes. 00065 mtrie_t subscriptions; 00066 00067 // Distributor of messages holding the list of outbound pipes. 00068 dist_t dist; 00069 00070 // True if we are in the middle of sending a multi-part message. 00071 bool more; 00072 00073 // List of pending (un)subscriptions, ie. those that were already 00074 // applied to the trie, but not yet received by the user. 00075 typedef std::basic_string <unsigned char> blob_t; 00076 typedef std::deque <blob_t> pending_t; 00077 pending_t pending; 00078 00079 xpub_t (const xpub_t&); 00080 const xpub_t &operator = (const xpub_t&); 00081 }; 00082 00083 class xpub_session_t : public session_base_t 00084 { 00085 public: 00086 00087 xpub_session_t (class io_thread_t *io_thread_, bool connect_, 00088 class socket_base_t *socket_, const options_t &options_, 00089 const char *protocol_, const char *address_); 00090 ~xpub_session_t (); 00091 00092 private: 00093 00094 xpub_session_t (const xpub_session_t&); 00095 const xpub_session_t &operator = (const xpub_session_t&); 00096 }; 00097 00098 } 00099 00100 #endif