ServerCore/Socket.cpp

#include "Socket.h"
#include "EPoll.h"
#include "Exception.h"
#include "Log.h"
#include "ZString.h"

namespace core
{

   void sigpipe_handler(int unused) {}

   Socket::Socket(EPoll &ePoll, std::string text) : ePoll(ePoll), text(text) {
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Socket object created [" << text << "].";
      buffer = (char *)malloc(4096);
      length = 4096;
   }

   Socket::~Socket() {
      shutDown = true;
      onUnregister();
      disableSocket();
      coreutils::Log(coreutils::LOG_DEBUG_4) << "Free on socket " << descriptor;
      free(buffer);
      if(descriptor == -1)
		  return;
      close(descriptor);
      coreutils::Log(coreutils::LOG_DEBUG_1) << text << " has ended (" << descriptor << ").";
   }

   void Socket::setDescriptor(int descriptor)
   {
      if ((descriptor == -1) && (errno == 24))
      {
         shutdown("Too many files open");
         throw coreutils::Exception("Too many files open. Refusing connection.");
      }
      coreutils::Log(coreutils::LOG_DEBUG_1) << text << " has started (" << descriptor << ").";
      if(descriptor < 3)
        throw coreutils::Exception("Descriptor out of range", __FILE__, __LINE__);
      this->descriptor = descriptor;
      onRegister();
      enableSocket();
      onRegistered();
   }

   int Socket::getDescriptor()
   {
      return descriptor;
   }

   void Socket::setBufferSize(int length)
   {
      this->length = length;
      buffer = (char *)realloc(buffer, length);
   }

   int Socket::getBufferSize()
   {
      return length;
   }

   void Socket::onRegister() {}

   void Socket::onRegistered() {}

   void Socket::onUnregister() {}

   void Socket::onUnregistered() {}

   bool Socket::eventReceived(struct epoll_event event, long long eventId) {
//      coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process beginning for socket " << getDescriptor();
      if(inHandler)
 //       coreutils::Log(coreutils::LOG_DEBUG_2) << "inHandler was already true.";
		inHandler = true;
      if(event.events & EPOLLRDHUP) {
//         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLRDHUP";
         readHangup = true;
         shutdown("hangup received");
      }
      if(event.events & EPOLLIN) {
//         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLIN";
         coreutils::ZString zbuffer(buffer, length);
         lock.lock();
         receiveData(zbuffer);
         if(!shutDown) {
	    inHandler = false;
         lock.unlock();
            resetSocket();
         }
      }
      if(event.events & EPOLLWRNORM) {
//         coreutils::Log(coreutils::LOG_DEBUG_2) << "EPOLLWRNORM";
	 writeSocket();
	 inHandler = false;
	   resetSocket();
      }
      inHandler = false;
//      coreutils::Log(coreutils::LOG_DEBUG_1) << "Event process ending for socket " << getDescriptor();
      return !shutDown;
   }
   
   void Socket::onDataReceived(std::string data)
     {
      throw coreutils::Exception("Need to override onDataReceived.", __FILE__, __LINE__, -1);
   }

   void Socket::onDataReceived(coreutils::ZString &data)
   {
      onDataReceived(std::string(data.getData(), data.getLength()));
   }

   void Socket::receiveData(coreutils::ZString &buffer) {
      coreutils::ZString blank("");
      if(buffer.getLength() <= 0)
        throw coreutils::Exception("Request to receive data with a zero buffer length.", __FILE__, __LINE__, -1);
      int len;
      int error = -1;
      if((len = ::read(getDescriptor(), buffer.getData(), buffer.getLength())) >= 0) {
         coreutils::ZString zbuffer(buffer.getData(), len);
//         coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer;
         onDataReceived(zbuffer);
      }
      else
      {

         error = errno;

         switch (error)
         {

            // When a listening socket receives a connection
            // request we get one of these.
            //
         case ENOTCONN:
            onDataReceived(blank);
            break;

         case ECONNRESET:
            break;

         default:
            throw coreutils::Exception("Error in read of data from socket.", __FILE__, __LINE__, error);
         }
      }
   }

   void Socket::writeSocket() {
      outlock.lock();
//      coreutils::Log(coreutils::LOG_DEBUG_3) << "writing data to socket " << getDescriptor();
      if(fifo.size() > 0) {
	 if(!shutDown)
	   int rc = ::write(descriptor, fifo.front().c_str(), fifo.front().length());
         fifo.pop();
      }
      outlock.unlock();
   }

   int Socket::write(std::string data) {
      outlock.lock();
      fifo.emplace(data);
      outlock.unlock();
      if(lock.try_lock()) {
         resetSocket();
         lock.unlock();
      }
      return 1;
   }

   void Socket::output(std::stringstream &out)
   {
      out << "|" << descriptor << "|";
   }

   bool Socket::needsToWrite()
   {
      return fifo.size() > 0;
   }

   void Socket::shutdown(std::string text) {
      coreutils::Log(coreutils::LOG_DEBUG_2) << "Shutdown requested from " << this->text << " (" << descriptor << ") with reason: " << text << ".";
      shutDown = true;
      reset = false;
   }

    void Socket::enableSocket() {
      struct epoll_event event;
      event.data.ptr = this;
      event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_ADD, getDescriptor(), &event);
   }

   void Socket::disableSocket() {
      epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_DEL, getDescriptor(), NULL);
   }

   void Socket::resetSocket() {
      struct epoll_event event;
      event.data.ptr = this;
      event.events = EPOLLIN | EPOLLRDHUP | EPOLLONESHOT | EPOLLET;
      if(fifo.size() > 0)
        event.events |= EPOLLWRNORM;
      if(!shutDown)
        epoll_ctl(ePoll.getDescriptor(), EPOLL_CTL_MOD, getDescriptor(), &event);
   }


}