JLigier.cc File Reference

ControlHost server. More...

#include <iostream>
#include <iomanip>
#include <string>
#include <sstream>
#include <deque>
#include <set>
#include "Jeep/JParser.hh"
#include "Jeep/JMessage.hh"
#include "JLang/JMemory.hh"
#include "JLang/JSharedCounter.hh"
#include "JNet/JSocket.hh"
#include "JNet/JSocketChannel.hh"
#include "JNet/JServerSocket.hh"
#include "JNet/JSelect.hh"
#include "JNet/JControlHost.hh"
#include "JSystem/JSystemToolkit.hh"
#include "JMath/JConstants.hh"

Go to the source code of this file.

Classes
class	JNET::JMemory_t< T >
class	JNET::JDispatch
	Data structure of a ControlHost message. More...
class	JNET::JClient
	ControlHost client manager. More...
class	JNET::JClientList
	List of ControlHost client managers. More...

Namespaces
	JNET
	Interprocess communication.

Typedefs
typedef JPrefix	JNET::JPrefix_t
typedef JSocketInputChannel < JPrefix_t >	JNET::JSocketInputChannel_t

Functions
int	JNET::getSizeOfPacket (const JPrefix_t &prefix)
	Get total size of internet packet. More...
void	JNET::setSizeOfPacket (const int size, JPrefix_t &prefix)
	Set total size of internet packet. More...
std::ostream &	JNET::operator<< (std::ostream &out, const JDispatch &message)
	Print message. More...
std::ostream &	JNET::operator<< (std::ostream &out, const JSocket &socket)
	Print socket. More...
std::ostream &	JNET::operator<< (std::ostream &out, const JSocketStatus &status)
	Print socket status. More...
std::ostream &	JNET::operator<< (std::ostream &out, const JSocketInputBuffer &buffer)
	Print socket input buffer. More...
int	main (int argc, char *argv[])

Detailed Description

ControlHost server.

Author

mdejong

Definition in file JLigier.cc.

Function Documentation

int main	(	int	argc,
		char *	argv[]
	)

Definition at line 579 of file JLigier.cc.

{
  using namespace std;
  using namespace JPP;

  int   port;
  int   backlog;
  int   timeout_us;
  int   buffer_size;
  int   debug;

  try {

    JParser<> zap("ControlHost server.");

    zap['P'] = make_field(port)                    = DISPATCH_PORT;
    zap['q'] = make_field(backlog)                 =   1024;
    zap['T'] = make_field(timeout_us)              =      1;
    zap['s'] = make_field(buffer_size)             = GIGABYTE;
    zap['Q'] = make_field(JClient::QUEUE_LIMIT)    =    100;
    zap['M'] = make_field(JDispatch::MEMORY_LIMIT) = (JSYSTEM::getRAM() >> 1);
    zap['d'] = make_field(debug)                   =      0;

    zap(argc, argv);
  }
  catch(const exception &error) {
    FATAL(error.what() << endl);
  }


  JServerSocket         server(port, backlog);
  JSelect               select;
  JClientList           clientList;


  DEBUG("Port         " << setw(10) << port                    << endl);
  DEBUG("Memory limit " << setw(10) << JDispatch::MEMORY_LIMIT << endl);
  DEBUG("Queue  limit " << setw(10) << JClient::QUEUE_LIMIT    << endl);


  for ( ; ; ) {

    select.reset();
    select.setReaderMask(server);

    for (JClientList::iterator client = clientList.begin(); client != clientList.end(); ++client) {

      if (!client->in.isReady()) {
        select.setReaderMask(*client);
      }

      if (client->out.isReset()) {

        if (!client->queue.empty()) {
        
          DEBUG("Client" << *client << ".set" << client->queue.front() << endl);
        
          client->out.set(client->queue.front());
          client->decrementRequest();
        
          select.setWriterMask(*client);
        }

      } else if (client->out.isBusy()) {

        select.setWriterMask(*client);
      }
    }

    if (select(timeout_us) > 0) {

      for (JClientList::iterator client = clientList.begin(); client != clientList.end(); ) {
        
        try {

          if (select.hasReaderMask(*client)) {

            try {
              client->in.read();
            }
            catch(const exception& error) {

              ERROR("Remove (3) client" << *client << "<" << client->getNickname() << ">: " << error.what() << endl);

              if (client->getNickname() != "") {
                clientList.add(JDispatch(DISPTAG_Died, client->getNickname()));
              }

              client->shutdown();

              client = clientList.erase(client);
              
              continue;
            }
              
            DEBUG("Client" << *client << ".read" << static_cast<const JSocketInputBuffer&>(client->in) << endl);
          }
                
          if (client->in.isReady()) {

            DEBUG("Message" << *client << ' ' << client->in.prefix.c_str() << ' ' << client->in.size() << endl);

            bool special = JControlHost::maybe_special(client->in.prefix);

            if (special) {

              client->in.seekg(sizeof(JPrefix_t));     // skip prefix
          
              if        (client->in.prefix.getTag() == DISPTAG_Subscribe) {

                client->setSubscription(string(client->in.getRemainingData(), client->in.getRemainingSize()));
                
              } else if (client->in.prefix.getTag() == DISPTAG_MyId) {
                
                client->setNickname(string(client->in.getRemainingData(), client->in.getRemainingSize()));
                
                clientList.add(JDispatch(DISPTAG_Born, client->getNickname()));
                
              } else if (client->in.prefix.getTag() == DISPTAG_Gime) {
                
                client->incrementRequest();
                
              } else if (client->in.prefix.getTag() == DISPTAG_Always) {
                
                client->setRequestAll();
                
              } else if (client->in.prefix.getTag() == DISPTAG_WhereIs) {
                
                string nick_name(client->in.getRemainingData(), client->in.getRemainingSize());
                string buffer;
                
                for (JClientList::iterator i = clientList.begin(); i != clientList.end(); ++i) {
                  if (i->getNickname() == nick_name) {
                    buffer += " " + i->getHostname();
                  }
                }
                
                JControlHost socket(*client);
                
                socket.PutFullString(DISPTAG_WhereIs, buffer.substr(buffer.empty() ? 0 : 1));
                
                DEBUG("Remove (1) client" << *client << endl);
                
                client->shutdown();
                
                client = clientList.erase(client);
                
                continue;         // skip any action

              } else if (client->in.prefix.getTag() == DISPTAG_ShowStat) {

                client->shutdown();
                
                client = clientList.erase(client);

                for (JClientList::iterator i = clientList.begin(); i != clientList.end(); ++i) {

                  int total = 0;

                  for (std::deque<JDispatch>::const_iterator message = i->queue.begin(); message != i->queue.end(); ++message) {
                    total += message->size();
                  }

                  cout << "client[" << i->getFileDescriptor() << "] " << i->getNickname() << endl;
                  cout << "tag - all:";
                  for (std::set<JTag>::const_iterator tag = i->getSubscriptionAll().begin(); tag != i->getSubscriptionAll().end(); ++tag) {
                    cout << ' ' << *tag;
                  }
                  cout << endl;
                  cout << "tag - any:";
                  for (std::set<JTag>::const_iterator tag = i->getSubscriptionAny().begin(); tag != i->getSubscriptionAny().end(); ++tag) {
                    cout << ' ' << *tag;
                  }
                  cout << endl;
                  cout << "queue " << i->queue.size() << ' ' << total << "B" << endl;
                }
                
                continue;         // skip any action

              } else if (client->in.prefix.getTag() == DISPTAG_Debug) {

                istringstream is(string(client->in.getRemainingData(), client->in.getRemainingSize()));

                is >> debug;

              } else {

                special = false;  // not a reserved tag.
              }
            } 
            
            if (!special) {

              clientList.add(JDispatch(client->in.prefix, client->in.data()));

              if (JDispatch::MEMORY_TOTAL > JDispatch::MEMORY_LIMIT) {

                WARNING("Memory " << setw(10) << JDispatch::MEMORY_TOTAL << " > " << setw(10) << JDispatch::MEMORY_LIMIT << endl);

                clientList.drop();
              }
            }
            
            client->in.reset();
          }
          
          if (select.hasWriterMask(*client)) {
              
            client->out.write();
            
            DEBUG("Client" << *client << ".write" << static_cast<const JSocketStatus&>(client->out) << endl);
            
            if (client->out.isReady()) {
              client->out.reset();
              client->queue.pop_front();
            }
          }

          ++client;
        }
        catch(const exception& error) {

          DEBUG("Remove (2) client" << *client << "<" << client->getNickname() << ">: " << error.what() << endl);
          
          if (client->getNickname() != "") {
            clientList.add(JDispatch(DISPTAG_Died, client->getNickname()));
          }

          client->shutdown();

          client = clientList.erase(client);
        }
      }

      if (select.hasReaderMask(server)) {

        JSocket socket;
        
        socket.accept(server.getFileDescriptor());
        
        socket.setSendBufferSize   (buffer_size);
        socket.setReceiveBufferSize(buffer_size);
        
        socket.setKeepAlive  (true);
        socket.setNonBlocking(true);
        
        DEBUG("New client" << socket << endl);

        clientList.push_back(JClient(socket));
      } 
    }
  }
}