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