L:/channel/channel/include/Msg.h

Go to the documentation of this file.

// Copyright (c) 2005, 2006 Yigong Liu
// Permission to use, copy, modify, distribute and sell this software for any 
//     purpose is hereby granted without fee, provided that the above copyright 
//     notice appear in all copies and that both that copyright notice and this 
//     permission notice appear in supporting documentation.
// The author makes no representations about the 
//     suitability of this software for any purpose. It is provided "as is" 
//     without express or implied warranty.

#ifndef _MSG_H_
#define _MSG_H_

#include "ace/Log_Msg.h"
//for ACE_GUARD macros, need the following 2 headers
#include "ace/Synch.h"
#include "ace/OS.h"
#include "ace/CDR_Stream.h"


#include <BaseDef.h>
#include <LinearIdTrait.h>
#include <HierarchicalIdTrait.h>
#include <Marshaler.h>

namespace channel {


  typedef void (*MsgFreeCallback) (char * data);

  template <class T>
    void DefaultMsgFreeCallback(char *c)
    {
      T *t = (T*) c;
      delete t;
    }

  template <class T>
    void DefaultMsgArrayFreeCallback(char *a)
    {
      T *t = (T*) a;
      delete[] t;
    }

  void DefaultNoOpMsgFreeCallback(char *)
    {
    }

  template <class SynchPolicy>
  class MsgDataHolder {
  public:
    char *data_;
    int ref_count_; 
    typename SynchPolicy::MUTEX ref_count_lock_;
    MsgDataHolder (char *data) : data_(data), ref_count_(1) {}
    void set_data(char *d, MsgFreeCallback gobbler) {
      ACE_GUARD (typename SynchPolicy::MUTEX, guard, ref_count_lock_);
      if (gobbler != NULL)
        gobbler (data_);
      else 
        delete data_; 
      data_ = d;
    }
    void ref_incr(void) {
      ACE_GUARD (typename SynchPolicy::MUTEX, guard, ref_count_lock_);
      ref_count_++;
    }
    void ref_decr(void) {
      ACE_GUARD (typename SynchPolicy::MUTEX, guard, ref_count_lock_);
      ref_count_--;
      if (ref_count_ < 0) ref_count_ = 0;
    }
    void release (MsgFreeCallback gobbler) {
      ACE_GUARD (typename SynchPolicy::MUTEX, guard, ref_count_lock_);
      if (ref_count_ > 1) { 
        ref_count_--;
      } else {
        if (gobbler != NULL)
          gobbler (data_);
        else 
          delete data_; 
        delete this;
      }
    }
  private:
    ~MsgDataHolder() {};
  };

  class DefaultMarshaler {
  public:
    static int marshal(ACE_OutputCDR &cdr, const char *data, const int size)
      {
      cdr << ACE_CDR::ULong (size);
      cdr.write_char_array (data, size);
      return cdr.good_bit();
      }
    static int demarshal(ACE_InputCDR &cdr, char * &data, int &size, MsgFreeCallback &callback)
      {
      ACE_CDR::ULong sz;
      cdr >> sz;
      size = (int)sz;
      data = new char[size];
      cdr.read_char_array(data, size);
      callback = DefaultMsgArrayFreeCallback<char>;
      return cdr.good_bit();
      }
  };    

  template <class IdType, class SynchPolicy>
  struct  Msg {    
    IdType type;
    int size_;
    MsgDataHolder<SynchPolicy> *data_holder_;
    MsgFreeCallback gobbler_;
    
    Msg () {}

    template <class MsgType>
    Msg (IdType t, MsgType *d, int s = sizeof(MsgType), 
         MsgFreeCallback g = DefaultMsgFreeCallback<MsgType>) : 
      type(t),size_(s),gobbler_(g) {
      data_holder_ = new MsgDataHolder<SynchPolicy>((char *)d);
    }

    Msg *clone (void) {
      Msg *m = new Msg();
      m->type = type;
      m->size_ = size_;
      m->data_holder_ = data_holder_;
      m->gobbler_ = gobbler_;
      m->data_holder_->ref_incr();
      return m;
    }
    ~Msg () {
      data_holder_->release(gobbler_);
    };

    char * data(void) {
      return data_holder_->data_;
    }
    template <class MsgType>
    void set_data(MsgType *d, int s = sizeof(MsgType), MsgFreeCallback g = DefaultMsgFreeCallback<MsgType>) {
      size_ = s;
      gobbler_ = g;
      data_holder_->set_data((char *)d, gobbler_);
    }

    int marshal(ACE_OutputCDR &cdr, Marshaler *mar) {
      if (IdTrait<IdType>::marshal(cdr, type) == 0) {
        return 0;
      }
      if (mar != NULL)
        mar->marshal(cdr, data_holder_->data_, size_);
      else
        DefaultMarshaler::marshal(cdr, data_holder_->data_, size_);
      return cdr.good_bit ();
    }

    int demarshal(ACE_InputCDR &cdr, Marshaler *mar) {
      //demarshal_IdType is already done at outside
      //IdTrait<IdType>::demarshal(cdr, type) == 0) {
      //demarshal data
      char *data;
      if (mar != NULL)
        mar->demarshal(cdr, data, size_, gobbler_);
      else
        DefaultMarshaler::demarshal(cdr, data, size_, gobbler_);
      data_holder_ = new MsgDataHolder<SynchPolicy>(data);
      return cdr.good_bit();
    }
      
  };



  struct Channel_Info_Msg {
    enum { MAX_ADDR_NAME_LEN = 128 };
    char host_addr[MAX_ADDR_NAME_LEN];
    u_short port;
    char unix_addr[MAX_ADDR_NAME_LEN];
    int conn_type;
    void *intf;  
    bool is_local;
    Channel_Info_Msg() {
      port = 0;
      host_addr[0] = '\0';
      unix_addr[0] = '\0';
      conn_type = 0;
      intf = NULL;
      is_local = false;
    }
  };

  template <class IdType>
  struct PubSub_Info_Msg {
    enum { MAX_REGISTRY_ENTRIES = 128 };
    int num_msg_types;
    IdType msg_types[MAX_REGISTRY_ENTRIES];
    PubSub_Info_Msg() {
      num_msg_types = 0;
    }
  };

  class ChannelMsg_Marshaler : public Marshaler {
  public:
    virtual int marshal(ACE_OutputCDR &cdr, const char *data, const int size) 
      {
        ACE_UNUSED_ARG(size);
        Channel_Info_Msg *msg = (Channel_Info_Msg *) data;
        int len = strlen(msg->host_addr);
        cdr << ACE_CDR::ULong (len);
        if (len > 0) 
          cdr.write_char_array (msg->host_addr, len);
        cdr << ACE_CDR::ULong (msg->port);
        len = strlen(msg->unix_addr);
        cdr << ACE_CDR::ULong (len);
        if (len > 0) 
          cdr.write_char_array (msg->unix_addr, len);
        cdr << ACE_CDR::ULong (msg->conn_type);
        return cdr.good_bit();
      }
    virtual int demarshal(ACE_InputCDR &cdr, char * &data, int &size, MsgFreeCallback &callback)
      {
        Channel_Info_Msg *msg = new Channel_Info_Msg();
        ACE_CDR::ULong num;
        cdr >> num;
        int len = (int)num;
        if (len > 0) {
          cdr.read_char_array(msg->host_addr, len);
          msg->host_addr[len] = '\0';
        }
        cdr >> num;
        msg->port = (u_short) num;
        cdr >> num;
        len = (int)num;
        if (len > 0) {
          cdr.read_char_array(msg->unix_addr, len);
          msg->unix_addr[len] = '\0';
        }
        cdr >> num;
        msg->conn_type = (int) num;
        data = (char*)msg;
        size = sizeof(Channel_Info_Msg);
        callback = DefaultMsgFreeCallback<Channel_Info_Msg>;
        return cdr.good_bit();
      }
  };

  template <class IdType>
  class PubSubMsg_Marshaler : public Marshaler {
  public:
    virtual int marshal(ACE_OutputCDR &cdr, const char *data, const int size) 
      {
        ACE_UNUSED_ARG(size);
        PubSub_Info_Msg<IdType> *msg = (PubSub_Info_Msg<IdType> *) data;
        cdr << ACE_CDR::Long (msg->num_msg_types);
        for (int i=0;i<msg->num_msg_types;i++) {
          IdType id = msg->msg_types[i];
          IdTrait<IdType>::marshal(cdr, id);
        }
        return cdr.good_bit();
      }
    virtual int demarshal(ACE_InputCDR &cdr, char * &data, int &size, MsgFreeCallback &callback)
      {
        PubSub_Info_Msg<IdType> *msg = new PubSub_Info_Msg<IdType>();
        ACE_CDR::Long num;
        cdr >> num;
        msg->num_msg_types = (int)num;
        for (int i=0;i<msg->num_msg_types;i++) {
          IdType id;
          IdTrait<IdType>::demarshal(cdr, id);
          msg->msg_types[i] = id;
        }
        data = (char*)msg;
        size = sizeof(PubSub_Info_Msg<IdType>);
        callback = DefaultMsgFreeCallback<PubSub_Info_Msg<IdType> >;
        return cdr.good_bit();
      }
  };

};

#endif

---