Mailbox.cc

Go to the documentation of this file.

/*
 *
 * Copyright 2006,2007 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.
 */

#include <ssrc/spread/Mailbox.h>
#include <fcntl.h>

__BEGIN_NS_SSRC_SPREAD

const Timeout Mailbox::ZeroTimeout(0, 0);


Mailbox::Mailbox(const string & connection, const string & name,
                 const bool group_membership, const Timeout & timeout,
                 const Priority priority)
  SSRC_DECL_THROW(Error)
{
  int result;
  Spread::group_type private_group;

  result =
    Spread::SP_connect_timeout(connection.c_str(),
                               (name.size() == 0 ? 0 : name.c_str()), priority,
                               group_membership, &_mbox, private_group,
                               timeout);

  if(result != Error::AcceptSession)
    throw Error(result);

  ::fcntl(_mbox, F_SETFD, (::fcntl(_mbox, F_GETFD) | FD_CLOEXEC));

  _connection = connection;
  _group_membership = group_membership;
  _private_group = private_group;
  _name = split_private_group(_private_group).first;

  _drop_receive  = false;
  _killed        = false;
}

int Mailbox::send(const ScatterMessage & message, const GroupList & groups)
  SSRC_DECL_THROW(Error)
{
  int result;

  result =
    Spread::SP_multigroup_scat_multicast(_mbox, message.service(),
                                         groups.size(), groups.groups(),
                                         message.type(), message.scatter());
  if(result < 0)
    throw Error(result);

  return result;
}

int Mailbox::send(const Message & message, const GroupList & groups)
  SSRC_DECL_THROW(Error)
{
  clear_message_parts();
  add_message_part(message);
  _scatter.set_type(message.type());
  _scatter.set_service(message.service());
  return send(_scatter, groups);
}

int Mailbox::send(const Message & message, const string & group)
  SSRC_DECL_THROW(Error)
{
  clear_groups();
  add_group(group);
  clear_message_parts();
  add_message_part(message);
  _scatter.set_type(message.type());
  _scatter.set_service(message.service());
  return send(_scatter, _groups);
}

int Mailbox::receive(ScatterMessage & message, GroupList & groups)
  SSRC_DECL_THROW(BufferSizeError, Error)
{
  int result, num_groups, endian_mismatch;
  BaseMessage::message_type type;
  BaseMessage::service_type stype;
  Spread::group_type sender;

  groups.resize(groups.capacity());
  message.init_pre_receive();

 try_again:
  type = 0, num_groups = 0, endian_mismatch = 0;
  stype = (_drop_receive ? BaseMessage::DropReceive : 0);

  result =
    Spread::SP_scat_receive(_mbox, &stype, sender, groups.size(), &num_groups,
                            groups.groups(), &type, &endian_mismatch,
                            message.scatter());
  if(!_drop_receive) {
    if(result == Error::GroupsTooShort) {
      if(num_groups < 0) {
        groups.resize(-num_groups);
        if(endian_mismatch == 0)
          goto try_again;
        else
          result = Error::BufferTooShort;
      } else
        throw BufferSizeError(Error::GroupsTooShort, -num_groups);
    }

    if(result == Error::BufferTooShort) {
      if(num_groups < 0)
        groups.resize(-num_groups);

      if(message.count_message_objects() > 0 && endian_mismatch < 0) {
        const unsigned int last_message = message.count_message_objects() - 1;
        Message *m = message.message(last_message);
        // endian_mismatch stores negative of incoming message size
        message.resize_message(last_message,
                               - endian_mismatch - message.size() + m->size());
        goto try_again;
      } else
        throw BufferSizeError(Error::BufferTooShort, -endian_mismatch);
    } else if(result < 0)
      throw Error(result);
  } else if(result < 0 && result != Error::GroupsTooShort &&
            result != Error::BufferTooShort)
    throw Error(result);

  message.set_type(type);
  message.set_service(stype);
  message.set_sender(sender);
  message.set_endian_mismatch(endian_mismatch != 0);

  groups.resize(num_groups);
  message.init_post_receive(result);

  return result;
}

__END_NS_SSRC_SPREAD