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