AsyncIO.h

#ifndef UTILS_ASYNC_IO
#define UTILS_ASYNC_IO

#include <string>
#include <vector>
#include <list>
#include <memory>
#include <string.h>
#include <KrisLibrary/Logger.h>
#include <KrisLibrary/File.h>
#include <KrisLibrary/Timer.h>
#include <KrisLibrary/errors.h>
#include <KrisLibrary/math/math.h>
#include "threadutils.h"

class AsyncReaderQueue
{
 public:
  AsyncReaderQueue(size_t queueMax=1000);
  virtual ~AsyncReaderQueue() {}
  void OnRead(const std::string& msg);
  void OnRead_NoLock(const std::string& msg);

  virtual void Reset();
  virtual void Work() {}
  int MessageCount() { return (int)msgCount; }
  int UnreadCount();
  std::string PeekNewest();
  std::string Newest() { msgQueue.clear(); return PeekNewest(); }
  std::vector<std::string> New();

  Mutex mutex;
  size_t queueMax;
  size_t msgCount;
  std::string msgLast;
  std::list<std::string> msgQueue;
  size_t numDroppedMsgs;
};

class AsyncWriterQueue 
{
 public:
  AsyncWriterQueue(size_t queueMax=1000);
  virtual ~AsyncWriterQueue() {}

  bool WriteAvailable() const { return !msgQueue.empty(); }
  std::string OnWrite();
  std::string OnWrite_NoLock();

  virtual void Reset();
  virtual void Work() {}
  void Send(const std::string& msg);
  int SentCount() { return msgCount+msgQueue.size(); }
  int DeliveredCount() { return (int)msgCount; }

  Mutex mutex;
  size_t queueMax;
  size_t msgCount;
  std::list<std::string> msgQueue;
  size_t numDroppedMsgs;
};

class AsyncPipeQueue
{
 public:
  AsyncPipeQueue(size_t recvQueueSize=1000,size_t writeQueueSize=1000);
  virtual void Reset() { reader.Reset(); writer.Reset(); }
  virtual void Work() {}

  void OnRead(const std::string& msg) { return reader.OnRead(msg); }
  void OnRead_NoLock(const std::string& msg) { return reader.OnRead_NoLock(msg); }
  bool WriteAvailable() const { return writer.WriteAvailable(); }
  std::string OnWrite() { return writer.OnWrite(); }
  std::string OnWrite_NoLock() { return writer.OnWrite_NoLock(); }

  int MessageCount() { return reader.MessageCount(); }
  int UnreadCount() { return reader.UnreadCount(); }
  std::string PeekNewest() { return reader.PeekNewest(); }
  std::vector<std::string> New() { return reader.New(); }
  std::string Newest() { return reader.Newest(); }

  void Send(const std::string& msg) { writer.Send(msg); }
  int SentCount() { return writer.SentCount(); }
  int DeliveredCount() { return writer.DeliveredCount(); }

  AsyncReaderQueue reader;
  AsyncWriterQueue writer;
};

class TransportBase
{
 public:
  TransportBase() {}
  virtual ~TransportBase() {}
  virtual bool Start() { return true; }
  virtual bool Stop() { return true; }
  virtual bool ReadReady() { return true; }
  virtual bool WriteReady() { return true; }
  virtual const std::string* DoRead() { return NULL; }
  virtual bool DoWrite(const char* msg,int length) { return false; }

  virtual bool DoWrite(const char* msg) { return DoWrite(msg,strlen(msg)); }
  virtual bool DoWrite(const std::string& msg) { return DoWrite(msg.c_str(),msg.length()); }
};


class StreamTransport : public TransportBase
{
 public:
  std::istream* in;
  std::ostream* out;
  enum Format { IntLengthPrepended, NullTerminated, Ascii, Base64 };
  Format format;
  std::string buffer;

  StreamTransport(std::istream& in);
  StreamTransport(std::ostream& out);
  StreamTransport(std::istream& in,std::ostream& out);
  virtual ~StreamTransport() {}
  virtual bool ReadReady();
  virtual bool WriteReady();
  virtual const std::string* DoRead();
  virtual bool DoWrite(const char* msg,int length);
};


class SocketClientTransport : public TransportBase
{
 public:
  SocketClientTransport(const char* addr);
  SocketClientTransport(const char* addr,SOCKET socket);
  virtual bool ReadReady();
  virtual bool WriteReady();
  virtual bool Start();
  virtual bool Stop();
  virtual const std::string* DoRead();
  virtual bool DoWrite (const char* str,int length);

  std::string addr;
  File socket;
  Mutex mutex;
  std::string buf;
};

class SocketServerTransport : public TransportBase
{
 public:
  SocketServerTransport(const char* addr,int maxclients=1);
  ~SocketServerTransport();
  virtual bool Start();
  virtual bool Stop();
  virtual bool ReadReady();
  virtual bool WriteReady();
  virtual const std::string* DoRead();
  virtual bool DoWrite (const char* str,int length);

  std::string addr;
  int serversocket;
  int maxclients;
  Mutex mutex;
  std::vector<std::unique_ptr<File> > clientsockets;
  int currentclient;
  std::string buf;
};

class SyncPipe : public AsyncPipeQueue
{
 public:
  SyncPipe();
  virtual ~SyncPipe();
  virtual void Reset();
  virtual void Work();
  virtual bool Start();
  virtual void Stop();
  inline bool Connected() { return initialized && (transport->WriteReady() || transport->ReadReady()); }
  inline bool WriteReady() { return initialized && transport->WriteReady(); }
  inline bool ReadReady() { return initialized && transport->ReadReady(); }

  std::unique_ptr<TransportBase> transport;

  bool initialized;
  Timer timer;
  double lastReadTime,lastWriteTime;
};


class AsyncReaderThread : public AsyncReaderQueue
{
 public:
  AsyncReaderThread(double timeout=Math::Inf);
  virtual ~AsyncReaderThread();
  virtual void Reset();
  virtual void Work() { LOG4CXX_ERROR(KrisLibrary::logger(),"No need to call Work on AsyncReaderThread"); }
  virtual bool Start();
  virtual void Stop();

  std::unique_ptr<TransportBase> transport;

  bool initialized;
  Thread thread;
  double timeout;

  Timer timer;
  double lastReadTime;
};

class AsyncPipeThread : public AsyncPipeQueue
{
 public:
  AsyncPipeThread(double timeout=Math::Inf);
  virtual ~AsyncPipeThread();
  virtual void Reset();
    virtual void Work() { LOG4CXX_ERROR(KrisLibrary::logger(),"No need to call Work on AsyncReaderThread"); }
  virtual bool Start();
  virtual void Stop();
  inline bool Connected() { return initialized && (transport->WriteReady() || transport->ReadReady()); }
  inline bool WriteReady() { return initialized && transport->WriteReady(); }
  inline bool ReadReady() { return initialized && transport->ReadReady(); }

  std::unique_ptr<TransportBase> transport;

  bool initialized;
  Thread workerThread;
  double timeout;

  Timer timer;
  Mutex mutex;
  double lastReadTime;
  double lastWriteTime;
};


class SocketReadWorker : public AsyncReaderThread
{
 public:
  SocketReadWorker(const char* addr,bool asServer=false,double timeout=Math::Inf)
    :AsyncReaderThread(timeout)
  {
    if(asServer)
      transport.reset(new SocketServerTransport(addr));
    else
      transport.reset(new SocketClientTransport(addr));
  }
};

class SocketPipeWorker : public AsyncPipeThread
{
 public:
  SocketPipeWorker(const char* addr,bool asServer=false,double timeout=Math::Inf)
    :AsyncPipeThread(timeout)
  {
    if(asServer)
      transport.reset(new SocketServerTransport(addr));
    else
      transport.reset(new SocketClientTransport(addr));
  }

  SocketPipeWorker(const char* addr,SOCKET socket,double timeout=Math::Inf)
    :AsyncPipeThread(timeout)
  {
    transport.reset(new SocketClientTransport(addr,socket));
  }
};

#endif