JControlHost.hh

Go to the documentation of this file.

#ifndef __JNET__JCONTROLHOST__
#define __JNET__JCONTROLHOST__

#include <string>
#include <sstream>
#include <arpa/inet.h>
#include <set>

#include "JNet/JPrefix.hh"
#include "JNet/JSocket.hh"
#include "JNet/JSocketBlocking.hh"
#include "JNet/JHostname.hh"
#include "JLang/JException.hh"
#include "JLang/JThrow.hh"
#include "JLang/JTimeval.hh"
#include "JMath/JMath.hh"


/**
 * \author mdejong
 */

namespace JNET {}
namespace JPP { using namespace JNET; }

namespace JNET {

  using JLANG::JControlHostException;
  using JLANG::JThrow;
  using JLANG::JTimeval;


  static const std::string   CHOO_VERSION  =  "1.0";
    

  /**
   * ControlHost subscription types.
   */
  enum JSubscription_t { 
    SUBSCRIBE_ALL            = 'a',
    SUBSCRIBE_ANY            = 'w'
    //SUBSCRIBE_SHARED_MEMORY  = 'm'
  };

  
  /**
   * ControlHost subscription.
   */
  class JSubscription :
    public JTag
  {
  public:
    /**
     * Constructor.
     *
     * \param  sub            subscription
     * \param  tag            tag
     */
    JSubscription(const JSubscription_t sub,
                  const JTag&           tag) :
      JTag        (tag),
      subscription(sub)
    {}


    /**
     * Get subscription type.
     *
     * \return                subscription
     */
    JSubscription_t getSubscription() const
    {
      return subscription;
    }
    
    
    /**
     * Convert subscription to string.
     *
     * \return                subscription
     */
    std::string toString() const 
    {
      return std::string(1, (char) subscription) + " " + JTag::toString();
    }


  protected:
    JSubscription_t subscription;
  };


  /**
   * Auxiliary class for all subscription.
   */
  struct JSubscriptionAll :
    public JSubscription
  {
    /**
     * Constructor.
     *
     * \param  tag            tag
     */
    JSubscriptionAll(const JTag& tag) :
      JSubscription(SUBSCRIBE_ALL, tag)
    {}
  };


  /**
   * Auxiliary class for any subscription.
   */
  struct JSubscriptionAny :
    public JSubscription
  {
    /**
     * Constructor.
     *
     * \param  tag            tag
     */
    JSubscriptionAny(const JTag& tag) :
      JSubscription(SUBSCRIBE_ANY, tag)
    {}
  };


  /**
   * Subscription list.
   */
  class JSubscriptionList :
    protected std::set<JSubscription> ,
    public JMATH::JMath<JSubscriptionList>
  {
  public:
    
    typedef std::set<JSubscription>::const_iterator            const_iterator;
    typedef std::set<JSubscription>::const_reverse_iterator    const_reverse_iterator;


    /**
     * Default constructor.
     */
    JSubscriptionList()
    {}


    /**
     * Copy constructor.
     *
     * \param  subscription   subscription
     */
    JSubscriptionList(const JSubscription& subscription)
    {
      add(subscription);
    }


    /**
     * Smart pointer.
     *
     * \return                pointer to set
     */
    const std::set<JSubscription>* operator->() const
    {
      return static_cast<const std::set<JSubscription>*>(this);
    }


    /**
     * Add subscription.
     *
     * \param  subscription   subscription
     */
    JSubscriptionList& add(const JSubscription& subscription) 
    {
      const_iterator p = this->find(subscription);

      if (p != this->end() && p->getID() == subscription.getID()) {

        if (p->          getSubscription() == SUBSCRIBE_ALL || 
            subscription.getSubscription() == SUBSCRIBE_ANY)
          return *this;        // maintain higher subscription level
        else
          this->erase(p);      // remove   lower  subscription level
      }

      this->insert(subscription);

      return *this;
    }    
   

    /**
     * Add subscription.
     *
     * \param  subscription   subscription
     * \return                this subscription
     */
    JSubscriptionList& add(const JSubscriptionList& subscription)
    {
      for (JSubscriptionList::const_iterator i = subscription.begin(); i != subscription.end(); ++i) {
        this->add(*i);
      }

      return *this;
    }


    /**
     * Convert subscription list to string.
     *
     * \return                subscription
     */
    std::string toString() const 
    {
      std::string buffer;

      for (const_iterator i = this->begin(); i != this->end(); ++i)
        buffer += ' ' + i->toString();

      return buffer;
    }
  };


  /**
   * Add operator.
   *
   * \param  first          subscription
   * \param  second         subscription
   * \return                subscription list
   */
  inline JSubscriptionList operator+(const JSubscription& first, const JSubscription& second)
  {
    JSubscriptionList buffer;

    buffer.add(first);
    buffer.add(second);

    return buffer;
  }


  /**
   * ControlHost class.
   */
  class JControlHost :
    public JSocketBlocking,
    public JThrow<JControlHost>
  {
  protected:
    /**
     * Default constructor.
     */
    JControlHost() :
      JSocketBlocking(JSocket(AF_INET, SOCK_STREAM))
    {
      configure();
    }


  public:
    /**
     * Check special ControlHost tags.
     *
     * \param  tag            tag
     * \return                true if possibly special tag; else false
     */
    static bool maybe_special(const JTag& tag) 
    {
      return tag[0] == '_';
    }


    /**
     * Check validity of subscription specifier.
     *
     * \param  c              subscription specifier
     * \return                true if valid; else false
     */
    static bool is_valid(const char c)
    {
      return (c == SUBSCRIBE_ALL ||
              c == SUBSCRIBE_ANY);// ||
              //c == SUBSCRIBE_SHARED_MEMORY);
    }
    

    /**
     * Constructor.
     *
     * \param  server         host name and optional port number
     */
    JControlHost(const JHostname& server) :
      JSocketBlocking(JSocket(AF_INET, SOCK_STREAM))
    {
      if (server.hostname != "")
        connect(server.hostname, server.port);
      else
        connect(server.port);

      configure();
    }
    

    /**
     * Constructor.
     *
     * \param  server         host name
     * \param  port           port
     */
    JControlHost(const std::string& server,
                 const int          port) :
      JSocketBlocking(JSocket(AF_INET, SOCK_STREAM))
    {
      connect(server, port);

      configure();
    }


    /**
     * Constructor.
     *
     * \param  ip_number      IP number
     * \param  port           port
     */
    JControlHost(const int ip_number,
                 const int port = DISPATCH_PORT) :
      JSocketBlocking(JSocket(AF_INET, SOCK_STREAM))
    {
      connect(ip_number, port);

      configure();
    }


    /**
     * Constructor.
     *
     * \param  socket         socket
     */
    JControlHost(const JSocket& socket) :
      JSocketBlocking(JSocket(socket))
    {}


    /**
     * Destructor.
     */
    ~JControlHost()
    {
      shutdown();
    }
    

    /**
     * Subscribe to single tag.
     *
     * \param  subscription   subscription
     * \return                0 if OK; -1 if socket error
     */
    int Subscribe(const JSubscription& subscription)
    {
      return PutFullString(DISPTAG_Subscribe, subscription.toString());
    } 


    /**
     * Subscribe to list of tags.
     *
     * \param  subscription   subscription
     * \return                0 if OK; -1 if socket error
     */
    int Subscribe(const JSubscriptionList& subscription)
    {
      return PutFullString(DISPTAG_Subscribe, subscription.toString());
    } 
   

    /**
     * Identify.
     *
     * \param  nick_name      nick name
     * \return                0 if OK; -1 if socket error
     */
    int MyId(const std::string& nick_name)
    {
      return PutFullString(DISPTAG_MyId, nick_name);
    } 
  
 
    /**
     * Tell server to send next message.
     *
     * \return                0 if OK; -1 if socket error
     */
    int SendMeNext()
    {
      return PutFullData(DISPTAG_Gime, NULL, 0);
    } 
   
 
    /**
     * Tell server to send messages forever.
     *
     * \return                0 if OK; -1 if socket error
     */
    int SendMeAlways()
    {
      return PutFullData(DISPTAG_Always, NULL, 0);
    } 
   
 
    /**
     * Send data.
     *
     * \param  tag            tag
     * \param  buffer         data
     * \param  length         number of bytes
     * \return                0 if OK; -1 if socket error; -2 if other error
     */
    int PutFullData(const JTag& tag, const void* buffer, const long long int length)
    {
      try {

        JPrefix __prefix__(tag, length);

        write((char*) &__prefix__, sizeof(JPrefix));

        if (length != 0) {
          write((char*) buffer, (int) length);
        }

        return 0;
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    }
   
 
    /**
     * Send data.
     *
     * \param  tag            tag
     * \param  buffer         data
     * \param  length         number of bytes
     * \return                0 if OK; -1 if socket error; -2 if other error
     */
    int PutFullData(const std::string& tag, const void* buffer, const long long int length)
    {
      try {
        return PutFullData(JTag(tag), buffer, length);
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
      catch (const JControlHostException& error) {
        return Throw(error, -2);
      }
    }


    /**
     * Send string.
     *
     * \param  tag            tag
     * \param  buffer         data
     * \return                0 if OK; -1 if socket error; -2 if other error
     */
    int PutFullString(const JTag&        tag, 
                      const std::string& buffer)
    {
      return PutFullData(tag, buffer.c_str(), buffer.size());
    }


    /**
     * Send string.
     *
     * \param  tag            tag
     * \param  buffer         data
     * \return                0 if OK; -1 if socket error; -2 if other error
     */
    int PutFullString(const std::string& tag, 
                      const std::string& buffer)
    {
      return PutFullData(tag, buffer.c_str(), buffer.size());
    }


    /**
     * Send version.
     *
     * \return                0 if OK; -1 if socket error; -2 if other error
     */
    int Connected()
    {
      return PutFullString(DISPTAG_Version, CHOO_VERSION);
    }
 
    
    /**
     * Wait for header.
     *
     * \param  prefix         prefix
     * \return                0 if OK; -1 if socket error
     */
    int WaitHead(JPrefix& prefix)
    {
      try {

        read((char*) &this->prefix, sizeof(JPrefix));
        
        prefix = this->prefix;

        return 0;
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    }
 
    
    /**
     * Wait for header.
     *
     * \param  tag            tag
     * \param  length         number of bytes
     * \return                0 if OK; -1 if socket error
     */
    int WaitHead(std::string& tag, long long int& length)
    {
      const int rvalue = WaitHead(this->prefix);

      if (rvalue == 0) {
        tag    = this->prefix.getTag();
        length = this->prefix.getSize();
      }

      return rvalue;
    }


    /**
     * Check for header, without waiting.
     *
     * \param  prefix         prefix
     * \param  timeout        timeout
     * \return                1 if header; 0 if no header; -1 if socket error
     */
    int CheckHead(JPrefix& prefix, JTimeval timeout = JTimeval::min())
    {
      try {

        if (in_avail(timeout)) {

          WaitHead(prefix);

          return 1;

        } else {

          return 0;
        }
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    }


    /**
     * Check for header, without waiting.
     *
     * \param  tag            tag
     * \param  length         number of bytes
     * \param  timeout_us     timeout [us]
     * \return                1 if header; 0 if no header; -1 if socket error
     */
    int CheckHead(std::string& tag, long long int& length, const int timeout_us = 0)
    {
      const int rvalue = CheckHead(this->prefix, timeout_us);

      if (rvalue == 1) {
        tag    = this->prefix.getTag();
        length = this->prefix.getSize();
      }

      return rvalue;
    }


    /**
     * Receive data.
     *
     * \param  buffer         data
     * \param  length         number of bytes
     * \return                0 if OK; -1 if socket error
     */
    int GetFullData(void* buffer, long long int length)
    {
      try {

        read((char*) buffer, (int) length);

        return 0;
      } 
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    }


    /**
     * Receive string.
     *
     * \param  buffer         data
     * \return                0 if OK; -1 if socket error
     */
    int GetFullString(std::string& buffer)
    {
      buffer.resize(this->prefix.getSize());
      
      return GetFullData((char*) buffer.data(), buffer.size());
    }


    /**
     * Locate ControlHost client(s).
     *
     * \param  host_name      host name
     * \param  nick_name      nick name
     * \param  answer         list of host names
     * \return                0 if OK; -1 if socket error
     */
    static int WhereIs(const std::string& host_name,
                       const std::string& nick_name, 
                       std::string&       answer)
    {
      try {

        using namespace std;

        JControlHost socket(host_name);
        
        socket.PutFullString(DISPTAG_WhereIs, nick_name);
        
        string        tag;
        long long int length;
        
        socket.WaitHead(tag, length);
        socket.GetFullString(answer);

        return 0;
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    } 
 

    /**
     * Configure socket (factory reset).
     *
     * \return                0 if OK; -1 if socket error
     */
    int configure()
    {
      try {

        setTcpNoDelay  (true);
        setReuseAddress(true);
        setKeepAlive   (true);
        setSendBufferSize   (128*1024);
        setReceiveBufferSize(128*1024);
        setNonBlocking (false);

        return 0;
      }
      catch (const JSocketException& error) {
        return Throw(error, -1);
      }
    } 


  private:
    mutable JPrefix prefix;

    /**
     */
    JControlHost(const JControlHost&);
    JControlHost& operator=(const JControlHost&);
  };


  /**
   * Match name.
   */
  typedef JControlHost   ControlHost;
}

#endif