ScatterMessage.h

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.
 */

#ifndef __SSRC_SPREAD_SCATTER_MESSAGE_H
#define __SSRC_SPREAD_SCATTER_MESSAGE_H

#include <ssrc/spread/Message.h>

#include <utility>
#include <vector>

__BEGIN_NS_SSRC_SPREAD

class ScatterMessage : public BaseMessage {
public:

  enum {
    MaxScatterElements = MAX_CLIENT_SCATTER_ELEMENTS
  };

private:
  friend class Mailbox;

  typedef std::pair<Message *, int> value_type;

  Spread::scatter _scatter;
  std::vector<value_type> _messages;
  unsigned int _size;

  const Spread::scatter *scatter() const {
    return &_scatter;
  }

  Spread::scatter *scatter() {
    return &_scatter;
  }

  void init_pre_receive();

  void init_post_receive(int bytes_received);

  Message * message(const unsigned int index) {
    return _messages[index].first;
  }

  void resize_message(const unsigned int message_index,
                      const unsigned int size)
  {
    value_type & v = _messages[message_index];
    Message *m = v.first;

    _size-=m->size();
    m->resize(size);
    _size+=m->size();
    _scatter.elements[v.second].buf = &((*m)[0]);
    _scatter.elements[v.second].len = m->size();
  }

protected:

#ifdef LIBSSRCSPREAD_ENABLE_MEMBERSHIP_INFO

  virtual int sp_get_membership_info(Spread::membership_info *info) const {
    return Spread::SP_scat_get_memb_info(&_scatter, service(), info);
  }

  virtual int sp_get_vs_set_members(const Spread::vs_set_info *vs_set,
                                        Spread::group_type member_names[],
                                        unsigned int member_names_count)
    const
  {
    return
      Spread::SP_scat_get_vs_set_members(&_scatter, vs_set, member_names,
                                         member_names_count);
  }

  virtual int sp_get_vs_sets_info(Spread::vs_set_info *vs_sets,
                                      unsigned int num_vs_sets,
                                      unsigned int *index)
    const
  {
    return
      Spread::SP_scat_get_vs_sets_info(&_scatter, vs_sets, num_vs_sets, index);
  }

#endif

public:

  // The initial vector capacity of 8 is an arbitrary choice.  We don't expect
  // a typical ScatterMessage to have too many parts.
  ScatterMessage() : _messages(8), _size(0) {
    _scatter.num_elements = 0;
  }

  virtual unsigned int size() const {
    return _size;
  }

  virtual void clear() {
    _scatter.num_elements = 0;
    _size = 0;
    _messages.clear();
  }

  unsigned int count_message_parts() const {
    return _scatter.num_elements;
  }

  unsigned int count_message_objects() const {
    return _messages.size();
  }

  bool add(const void *data, const unsigned int size);

  bool add(const Message & message) {
    if(add(&message[0], message.size())) {
      // We cheat and cast away constness!
      _messages.push_back(value_type(const_cast<Message*>(&message),
                                     _scatter.num_elements - 1));
      return true;
    } else
      return false;
  }
};


__END_NS_SSRC_SPREAD

#endif