Mailbox.h

Go to the documentation of this file.

  /* Copyright 2006 Savarese Software Research Corporation
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.savarese.com/software/ApacheLicense-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  #ifndef __SSRC_SPREAD_MAILBOX_H
  #define __SSRC_SPREAD_MAILBOX_H
  
  #include <ssrc/spread/ScatterMessage.h>
  
  __BEGIN_NS_SSRC_SPREAD
  
  class Timeout {
   Spread::sp_time time;
  
  public:
   Timeout(const Spread::sp_time time) : time(time) { }
  
   Timeout(const long sec = 0, const long usec = 0) {
   time.sec = sec, time.usec = usec;
   }
  
   operator Spread::sp_time() const { return time; }
  };
  
  class Mailbox {
  public:
   typedef Spread::mailbox descriptor_type;
  
   static const Timeout ZeroTimeout;
  
   enum Priority {
   Low = LOW_PRIORITY, Medium = MEDIUM_PRIORITY, High = HIGH_PRIORITY
   };
  
  private:
   bool _group_membership, _drop_receive, _killed;
   string _connection, _name, _private_group;
   descriptor_type _mbox;
   ScatterMessage _scatter;
   GroupList _groups;
  
  public:
  
   explicit Mailbox(const string & connection = "", const string & name = "",
   const bool group_membership = true,
   const Timeout & timeout = ZeroTimeout,
   const Priority priority = Medium) SSRC_DECL_THROW(Error);
  
   ~Mailbox() {
   if(!killed())
   Spread::SP_disconnect(_mbox);
   }
  
   const string & connection() const {
   return _connection;
   }
  
   const string & name() const {
   return _name;
   }
  
   descriptor_type descriptor() const {
   return _mbox;
   }
  
   const string & private_group() const {
   return _private_group;
   }
  
   bool group_membership() const {
   return _group_membership;
   }
  
   void set_drop_receive(const bool drop = true) {
   _drop_receive = drop;
   }
  
   bool drop_receive() const {
   return _drop_receive;
   }
  
   void join(const string & group) SSRC_DECL_THROW(Error) {
   int result = Spread::SP_join(_mbox, group.c_str());
   if(result != 0)
   throw Error(result);
   }
  
   void leave(const string & group) SSRC_DECL_THROW(Error) {
   int result = Spread::SP_leave(_mbox, group.c_str());
   if(result != 0)
   throw Error(result);
   }
  
  #ifdef LIBSSRCSPREAD_ENABLE_MAILBOX_KILL
  
   void kill() {
   Spread::SP_kill(_mbox);
   _killed = true;
   }
  #endif
  
   bool killed() const {
   return _killed;
   }
  
   unsigned int poll() const SSRC_DECL_THROW(Error) {
   int result = Spread::SP_poll(_mbox);
   if(result < 0)
   throw Error(result);
   return result;
   }
  
   bool add_message_part(const void *data, const unsigned int size) {
   return _scatter.add(data, size);
   }
  
   bool add_message_part(const Message & message) {
   return _scatter.add(message);
   }
  
   void add_group(const string & group) {
   _groups.add(group);
   }
  
   void add_groups(const GroupList & groups) {
   _groups.add(groups);
   }
  
   string group(const unsigned int index) const {
   return _groups[index];
   }
  
   void copy_groups(GroupList & groups) {
   groups = _groups;
   }
  
   unsigned int count_groups() const {
   return _groups.size();
   }
  
   void clear_groups() {
   _groups.clear();
   }
  
   unsigned int count_message_parts() const {
   return _scatter.count_message_parts();
   }
  
   void clear_message_parts() {
   _scatter.clear();
   }
  
   int send(const Message & message, const string & group)
   SSRC_DECL_THROW(Error);
  
   int send(const Message & message, const GroupList & groups)
   SSRC_DECL_THROW(Error);
  
   int send(const ScatterMessage & message, const GroupList & groups)
   SSRC_DECL_THROW(Error);
  
   int send(const ScatterMessage & message, const string & group)
   SSRC_DECL_THROW(Error)
   {
   clear_groups();
   add_group(group);
   return send(message, _groups);
   }
  
   int send(const GroupList & groups, const BaseMessage::message_type type = 0,
   const BaseMessage::service_type service = BaseMessage::Safe)
   SSRC_DECL_THROW(Error)
   {
   _scatter.set_type(type);
   _scatter.set_service(service);
   return send(_scatter, groups);
   }
  
   int send(const string & group, const BaseMessage::message_type type = 0,
   const BaseMessage::service_type service = BaseMessage::Safe)
   SSRC_DECL_THROW(Error)
   {
   clear_groups();
   add_group(group);
   return send(_groups, type, service);
   }
  
   int send(const BaseMessage::message_type type = 0,
   const BaseMessage::service_type service = BaseMessage::Safe)
   SSRC_DECL_THROW(Error)
   {
   return send(_groups, type, service);
   }
  
   int send(const Message & message) SSRC_DECL_THROW(Error) {
   return send(message, _groups);
   }
  
   int send(const ScatterMessage & message) SSRC_DECL_THROW(Error) {
   return send(message, _groups);
   }
  
   int receive(ScatterMessage & message, GroupList & groups)
   SSRC_DECL_THROW(BufferSizeError, Error);
  
   int receive(Message & message, GroupList & groups)
   SSRC_DECL_THROW(BufferSizeError, Error)
   {
   clear_message_parts();
   add_message_part(message);
   return receive(_scatter, groups);
   }
  
   int receive(GroupList & groups) SSRC_DECL_THROW(BufferSizeError, Error) {
   return receive(_scatter, groups);
   }
  
   int receive(Message & message) SSRC_DECL_THROW(BufferSizeError, Error) {
   return receive(message, _groups);
   }
  
   int receive(ScatterMessage & message)
   SSRC_DECL_THROW(BufferSizeError, Error)
   {
   return receive(message, _groups);
   }
  
   int receive() SSRC_DECL_THROW(BufferSizeError, Error) {
   return receive(_scatter, _groups);
   }
  
  };
  
  __END_NS_SSRC_SPREAD
  
  #endif