DRDE/AusRegCliever/include/mdcommon.h

  /**
   * \mainpage AusReg Cliever
   *
   * <br><hr>
   *
   * \par About
   * \par
   * <a href="http://dnseppus.meansofproduction.biz/doc/ausregcliever">The Generics Document</a>
   * 
   * \htmlonly
   * \endhtmlonly
   *
   * <br><hr>
   *
   * \par AusReg Cliever requirements
   *
   * The AusReg Cliever was originally developed on Debian Wheezy 64 but
   * should compile on any Unix where all of the requirements are present. The master
   * daemon programs was developed for a cluster but in this design that's cut back to
   * a stub.
   *
   *
   * <br><hr>
   *
   * \par Access AusReg Cliever source code
   *
   *
   *  See 'DEVOPS" at the delivert/support site
   *
   *
   *
   * <br><hr>
   *
   * \par Top Level External Dependencies... and version currently used 
   * <br>
   * \b boost:    1.49    <br>
   * \b log4cpp:  1.0     <br>
   * \b AusRegTk: 1.3.2   <br>
   * <br>
   * See the Tk for its' dependencies, besides that and above everything should be in the git tree.
   * 
   * <br><hr>
   *
   * mdcommon.h - Core Daemon Level definitions.
   *
   * \par Authors Support Site
   *
   * \htmlonly <a href=http://dnseppus.meansofproduction.biz/doc/drkoeln.de>Delivery/Suppor site</a> \endhtmlonly
   *
   * \par License
   * juan@acm.org for Secura
   *
   */

#ifndef MD_COMMON
#define MD_COMMON

#include<boost/asio/datagram_socket_service.hpp>

using namespace std;
using boost::asio::ip::udp;

#define MAX_SCENARIO                 1        // defined by increasing assumption of control from TK
#define MAX_AC                       1
#define MAX_CLIENTS                  1
#define MAX_CLIEVER                  10
#define MAX_PEER                     10
#define MAX_DEBUG                    1000     // Set the config parm higher than this to inhibit logNdebug(m...) where m < this
#define MAX_DEVICE                   MAX_PEER + MAX_CLIEVER + MAX_CLIENTS + 1
#define MD_EPOCH                     date()           
#define MD_HEARTBEAT                 1            // Network peer heartbeat in seconds.
#define MD_MAX_DATAGRAM              (32*1024)    // half of the IPV4 max
#define CLIEVER_APP                  "AusReg Cliever"
#define MD_COMPONENT                 "Master Daemon" // Cliever Component
#define MD_NAME                      CLIEVER_APP " " MD_COMPONENT
#define MD_VERSION                   " 1.0 "         // Version
#define MD_REFRESH                   10              // default milliseconds between API frame updates
#define MD_TYPE                      "EPPPEER"
#define NORMAL_DEBUG                 10  
#ifndef CURRENT_DEBUG
#define CURRENT_DEBUG                NORMAL_DEBUG
#endif
#define NOT_OK                       -1
#define OK                           0
#define RUNNING_DIR          	     "/tmp"

#define theMachine                   thisConfig->machine[thisConfig->thisMachineContext]

enum md_mand {
  MD_NAM,                            // Not a Mand.
  MD_USER_MAND,                      // User defined mand type
  MD_EPP,                            // RFC defined type
  MD_RULE_ACTION,
  MD_CLIEVER_CMD,                    // From CD command loop or other UI
  MD_CLIENT_CMD,                     // From APIG to MD
  MD_NMANDS 
};

enum mdErrorCode {                   // In auc-md/kb index to masterDaemonConfig.err
  MDERR_OK,
  MDERR_MISSING,
  MDERR_EXISTS,
  MDERR_CONFLICT,
  MDERR_NOTREADY,
  MDERR_SYNTAX,
  N_MDSTDERR
};

enum md_dispatch_category {
  MD_NEWBORN   = 0,                  // inbound this isn't dispatched
  CD_FRAME,
  MD_FRAME,
  DV_MDQUERY,
  MD_SHUTDOWN
};

enum md_device {
  MDDEV_MD = 0,
  MDDEV_CD,
  MDDEV_CLIENT,
  MDDEV_PEER,
  MACHINE,
  N_MDDEV_TYPES
};

enum md_units {
  MD_UNITS_UNDEFINED = 0,
  N_MD_UNITS
};

enum mdDGtype {
  MDDG_HEARTBEAT = 0,
  MDDG_DEVICEHB,
  MDDG_NEWBORN,
  MDDG_MDQUERY,
  MDDG_REGSCPI,
  MDDG_REGOBS,
  MDDG_REGODE,
 MDDG_TELEMETRY,
 MDDG_CDRESET,
  N_MDDG_TYPES
};

typedef
struct MD_DG_TYPE {
unsigned inBandOnly   :  1;
unsigned requiresAck  :  1;
unsigned isAckNak     :  1;
unsigned value        :  1; // i.e. boolean for ack/nak etc. where true = ack.
unsigned reserved     :  4;
unsigned clieverGroup :  8; // masterDaemonConfig.thisMachineContext
unsigned reserved2    : 16; 
}
 mdDGflags;

typedef 
struct MD_DG_HEADER {
mdDGtype    msgType;
mdDGflags   dgType;
mdDGtype    dgSubType;
int         msgSN;           // Ordinal in a dialog 
md_device   clientType;
int         sourceHandle;
int         sinkHandle;
int         handle;          // for example if query about a device other than source or sink.
mdErrorCode ec;
int         payloadSize;
int         primeOffset;     // for example to the name associated with the data
}
 mdDGHeader;

typedef
struct MD_DATAGRAM {
  mdDGHeader hdr;
  char       payLoad[MD_MAX_DATAGRAM - (sizeof(mdDGHeader))];
}
 mdDatagram;

typedef 
struct MD_REPLY {
 mdDGHeader hdr;
 char  payLoad[512 - sizeof(mdDGHeader)];
}
 mdDGReply;

class mdReply {
public:

 mdDGReply  dg;

 mdReply() {memset((void *)&dg,0,sizeof(mdDGReply));
            dg.hdr.dgType.isAckNak = true;
           }
};

#if (defined(MD_MAIN) )
const char *clientTypes[N_MDDEV_TYPES ] = { "Master Daemon",  "Cliever", "MD Client" };
#else
extern const char *clientTypes[N_MDDEV_TYPES];
#endif

/* \brief mdProcess
 *        Abstract base class of various subprocesses
 *
 *  AC is (potentially) a distributed system with various subprocesses
 *  spanning the central server, the 'Cliever' middleware component and clients.
 *  Root class for these.
 *
 */

class mdWQitem {
public:
 md_dispatch_category kind;
 const void          *what;
 int                  user; // User defined
 mdWQitem (const void *work,md_dispatch_category priority,int x)
 : what(work), kind(priority), user(x) {}
 
 bool operator< (const mdWQitem & right) const {
  return kind < right.kind; 
 }

};

typedef std::priority_queue < mdWQitem* > MDWQ;

class mdProcess {  
public: MDWQ q;
                   mdProcess()  {}
                  ~mdProcess()  {}

        void  queue(mdWQitem *w) {q.push(w);}             
virtual void  dispatch(mdWQitem *w)=0;
virtual void  run() = 0;  

};
#define MD_DEFAULT_DEVICE_PROTOCOL 0
#define MD_DEFAULT_RULE            0

typedef
struct { 
  unsigned       open:1;  
  unsigned     looped:1;  // back channel established from passive connection.     
  unsigned  reserved:30;
 } mdCnctBool;

typedef 
struct  MD_CONTROL_BLOCK
{int                                  handle;        // debug mark
 mdCnctBool                       connection;
 boost::asio::ip::udp::endpoint           ep;
 boost::asio::ip::udp::resolver::iterator it;
 boost::asio::ip::udp::socket           *udp;
 MD_CONTROL_BLOCK() {
  handle = 0;
  memset((void *)&connection,0,sizeof(connection));
  udp    = NULL;
 }
}
 mdCB;

typedef std::map<int,mdCB*> mdStdDevicePODMap;      // MD Standard Device Map -
/*
 *\brief stdDev: collection with       MD at: 0, CD in:  1 to MAX_CLIEVER,
 *                                     PEERS  in:  MAX_CLIEVER + 1 to - MAX_CLIEVER+1+MAX_PEER,
 * 
 * The mdStdDevIdx of a device is its index above. 
 * Within intervals clients are assigned compactly in the same order as thier handles are created and assigned.
 */

#if (defined(MD_MAIN) || defined(CD_MAIN) )
         mdStdDevicePODMap  cb;
         int                myStdDevIdx=MAX_CLIENTS+1; // global default to localhost
#else
  extern mdStdDevicePODMap  cb;
  extern int                myStdDevIdx;
#endif

class mdDGChannel
{public:

  boost::asio::io_service&                         io_service_;
  boost::asio::ip::udp::endpoint                   p_endpoint_;
  boost::asio::ip::udp::endpoint                   a_endpoint_;
  boost::asio::ip::udp::endpoint                          *ep_;
  boost::asio::ip::udp::resolver                            *r;
  boost::asio::ip::udp::resolver::query                     *q;
  boost::asio::ip::udp::socket                        passive_;
  boost::asio::ip::udp::socket                        *active_;
  boost::asio::ip::udp::socket                             *s_;
  
  char                                 ack_[sizeof(mdDGReply)];
  char                                  data_[MD_MAX_DATAGRAM];

  int                                              mdStdDevIdx;

  mdDatagram                                        *inProcess;
  mdReply                                               *reply;

  short                                                   port;

  mdDGChannel(boost::asio::io_service& io_service,
              short inport, int stdDevIdx=myStdDevIdx)
    : io_service_(io_service), mdStdDevIdx(stdDevIdx), 
      passive_(io_service, udp::endpoint(udp::v4(), inport))
  {   
   inProcess   = (mdDatagram *)data_;
   ep_         = &p_endpoint_;
   port        = inport;
   reply       = (mdReply *)ack_;
   q           = NULL;
   r           = NULL;
   s_          = &passive_;
   
    passive_.async_receive_from(
        boost::asio::buffer(data_, MD_MAX_DATAGRAM), p_endpoint_,
        boost::bind(&mdDGChannel::handle_receive_from, this,
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
  }

  void async_send(mdDatagram &dg) {

      size_t dgSize = sizeof(mdDGHeader) + dg.hdr.payloadSize;

      *inProcess = dg;
      //std::memcpy(this->data_,(char *)&dg,dgSize);

      active_->async_send_to(
          boost::asio::buffer(data_, dgSize), *ep_,
          boost::bind(&mdDGChannel::handle_send_to, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
   }

   bool connect_to (std::string &host, std::string &port, int stdDevIdx=-1) {

   bool value = false;
   int  rport = atoi(port.c_str());
   boost::system::error_code    ec; 
   boost::asio::ip::udp::endpoint remote ( boost::asio::ip::address::from_string(host.c_str()), rport);

   stdDevIdx = (stdDevIdx == -1) ? mdStdDevIdx : stdDevIdx;
   map<int,mdCB *>::iterator iter = cb.find(stdDevIdx);
   if( iter == cb.end() ) cb[stdDevIdx] = new mdCB();

   try {
     if (!r) r = new udp::resolver(io_service_);
     if ( q)  delete q;
             q = new udp::resolver::query(udp::v4(), host.c_str(), port.c_str());
     cb[stdDevIdx]->it = r->resolve(*q);
     if (!cb[mdStdDevIdx]->udp)
          cb[mdStdDevIdx]->udp = new udp::socket(io_service_, udp::endpoint(udp::v4(), 0));
     active_ = cb[mdStdDevIdx]->udp;
     if (active_->is_open())
         active_->connect(remote,ec);

     if (!ec) value = true;  

   }
    catch(...) {}

   return (cb[stdDevIdx]->connection.open=value);

  }

  bool connect_to (boost::asio::ip::udp::endpoint &ep,boost::system::error_code &ec,int &step,int stdDevIdx=-1)
  {
   
   bool value = false;

   stdDevIdx = stdDevIdx == -1 ? mdStdDevIdx : stdDevIdx;
   map<int,mdCB *>::iterator iter = cb.find(stdDevIdx);
    if( iter == cb.end() ) cb[stdDevIdx] = new mdCB();

   try { if (cb[stdDevIdx]->udp) {if (cb[stdDevIdx]->udp->is_open())
                                      cb[stdDevIdx]->udp->close();
                                  delete cb[stdDevIdx]->udp;}

      cb[stdDevIdx]->ep  = ep;
      active_ = cb[stdDevIdx]->udp = new boost::asio::ip::udp::socket( io_service_ , udp::endpoint(udp::v4(), 0) );
      ec.clear();
      active_->connect( cb[stdDevIdx]->ep, ec );
      if (active_->is_open()) { value = true; cb[stdDevIdx]->connection.open=1; }
      else {
         step++;
         active_->open( udp::v4(),  ec );
         if (active_->is_open()) {value = true; cb[stdDevIdx]->connection.open=1; }
      }

   } 
   catch(boost::system::system_error &be) {boost::system::system_error warning = be;} 
   catch(...) {}

   return (value);

  }

  void handle_receive_from(const boost::system::error_code& error, size_t bytes_recvd);

  void handle_send_to(const boost::system::error_code& asioEC,  size_t sentByes)
  {
     size_t dgSize = inProcess->hdr.dgType.requiresAck
                   ?  sizeof(mdReply) :  sizeof(mdDatagram);

      //  std::string debugMsg = asioEC.message();

      if (inProcess->hdr.dgType.requiresAck)
        active_->async_receive_from(
        boost::asio::buffer(ack_, dgSize), *ep_,
        boost::bind(&mdDGChannel::handle_receive_from, this,
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
      else
        active_->async_receive_from(
        boost::asio::buffer(data_, dgSize), *ep_,
        boost::bind(&mdDGChannel::handle_receive_from, this,
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
  }

  bool send(mdDatagram &dg) {

      size_t dgSize = sizeof(mdDGHeader) + dg.hdr.payloadSize;

      return (dgSize == 
      active_->send(boost::asio::buffer((void *)&dg, dgSize)));

  }

  bool send_to(mdDatagram &dg, int mdStdDevIdx) {

      size_t dgSize = sizeof(mdDGHeader) + dg.hdr.payloadSize;

      return (dgSize == 
      active_->send_to(boost::asio::buffer((void *)&dg, dgSize), *cb[mdStdDevIdx]->it));

  }

  size_t send(mdReply &dg, boost::system::error_code &ec) {

      size_t dgSize = sizeof(mdDGReply);
      boost::asio::socket_base::message_flags  flags=0;

      return (active_->send(boost::asio::buffer((void *)&dg, dgSize), flags, ec));

  }
 
  size_t send_back(mdReply &dg, boost::system::error_code &ec, int &step) {

      size_t dgSize = sizeof(mdDGReply);
      boost::asio::socket_base::message_flags  flags=0;
      step = 1;

      a_endpoint_ = passive_.remote_endpoint(ec);

      if (ec) return 0;
    
      return (passive_.send_to(boost::asio::buffer((void *)&dg, dgSize), a_endpoint_, flags, ec));

  }

  bool send_to(mdReply &dg,int mdStdDevIdx) {

      size_t dgSize = sizeof(mdDGReply);

      return (dgSize == 
      active_->send_to(boost::asio::buffer((void *)&dg, dgSize), *cb[mdStdDevIdx]->it));

  }

};

        
typedef std::map<int,std::string> mdErrMsgMap;

class mdError {
 
  int         instance; 
  mdErrMsgMap     text;

public: 
   mdError() { text[0]  = std::string("OK");
               instance = 0;
             }

  ~mdError() {}
 
   // The below populate the additional messages;
   // Instances below 1000 are reserved for system errors, negative
   // integers not used.
   // Users derive from this class and implement additionalUsrMsgs.

   void     additionalUserMsg(int instCode,const char *msgText)
   {if (instCode >= 1000) text[instCode] = std::string(msgText);}
   void additionalSystemMsg(int instCode,const char *msgText)
   {if (instCode < 1000 && instCode >0) text[instCode] = std::string(msgText);}

   int      get()        {return instance;}
   void     get(mdError **parentPtr)
                         {*parentPtr = this;}
   void     set(int i)   {instance = i;}

   const char *what(char *buffer=NULL) {
     if (text.find(get()) == text.end()) {
      {if (!buffer) return NULL;
       sprintf(buffer,"Unknown error code: %d",get());
       return buffer;
      } 
      return text[get()].c_str();
     }
     return "unknown";
   }

};

class mdDG {
public:
  mdDatagram dg;

  mdDG() {memset(&dg,0,sizeof(mdDatagram));}

};

#endif