#ifndef EDISTCOMP_H
#define EDISTCOMP_H


#include "estr.h"
#include "evar.h"
#include "efile.h"
#include "esocket.h"
#include "estrhash.h"
#include "einthashof.h"
#include "ethread.h"

#include <queue>
#include <list>

using namespace std;

class eatom_base;
class ecodeAtom;
//class edctask;

class evarRemote;

class edcnode : public eworker
{
 public:
  estr recvbuffer;
  estr sendbuffer;

  int oldoutfd;

  etaskQueue *tqueue;
  evar result;
  einthashof<etaskBase*> rTasks;
  int threads;
  int taskid;
//  etaskBase *rTask;
//  evararray rArgs;

//  einthashof<evar> tmpVars;
  long tmpVarNext;

  estr id;
  estr hostname;
  unsigned int pid;
  estr group;

  efunc onReady;

  bool isChoked;
  bool isBusy;
  bool showResult;

  emutex mutex;

  edcnode();
  virtual ~edcnode();

  virtual void doSend()=0;
  virtual void doRecv()=0;
  virtual void doClose()=0;
  virtual void doWait()=0;

  virtual bool sendMsg(int code,const estr& msg);

  void remoteFree(const evarRemote& rv);
  evar remoteMethod(const evarRemote& rv,const estr& mname,const evararray& args);
  evar remoteProperty(const evarRemote& rv,const estr& pname);
  evar remoteValue(const evarRemote& rv);

  void call(const estr& cmd,const evararray& arr); 
  void interpret(const estr& cmd); 
  earray<estr> autocomplete(const estr& rhost,const estr& cmd); 
  evar executeAtom(const estr& exechost,eatom_base *atom,estrhashof<evar>& env);
  evar executeCodeAtom(const estr& exechost,ecodeAtom *catom,estrhashof<evar>& env);

  virtual void execute(etaskBase& task,const efunc& func,const evararray& args);
  virtual void dispatch();

  void doProcess();
  void doHandleAutocomplete(const estr& acdata);
  void doHandleExecuteAtom(const estr& data);
  void doHandleExecuteCodeAtom(const estr& data);
  void doHandleRunTask(const estr& data);
  void doHandleCall(const estr& data);
  void doHandleEval(const estr& data);

  void doHandleOutput(const estr& data);
  void doHandleResult(const estr& data);
  void doHandleResult2(const estr& data);
  void doHandleTaskResult(const estr& data);
  void doHandleHostname(const estr& data);
  void doHandleHosts(const estr& data);

  void doHandleRemoteResult(const estr& resdata);

  void handleRemoteCreate(const estr& data);
  void handleRemoteFree(const estr& data);
  void handleRemoteMethod(const estr& data);
  void handleRemoteProperty(const estr& data);
  void handleRemoteValue(const estr& data);

  void sendResult(const evar& var);
  void sendRemoteResult(const estr& returnhost,int returnpid,long rvid);
  void sendResult(const estr& returnhost,int returnpid,const evar& var);
  void sendTaskResult(const estr& returnhost,int returnpid,int taskid,const evar& var);
  void sendHostname(const estr& group="");
  void sendHosts();
  void sendNewHosts(const earray<estr>& newhosts);
};

class edcnode_proc : public edcnode
{
 public:
  int outpipe;
  efile fin;
  efile fout;
  esystemCallback *outCallback;

  edcnode_proc();
  edcnode_proc(int infd,int outfd);

  virtual void doSend();
  virtual void doRecv();
  virtual void doClose();
  virtual void doWait();

  void sendOutput();
};

#if !defined(_WIN32)
class edcnode_unix : public edcnode,public enamedsocket
{
 public:
  edcnode_unix();
  edcnode_unix(enamedsocket& s);

  void doConnected();

  virtual void doSend();
  virtual void doRecv();
  virtual void doClose();
  virtual void doWait();
};
#endif

class edcnode_tcp : public edcnode,public esocket
{
 public:
  edcnode_tcp();
  edcnode_tcp(esocket& s);

  int handshake;
  estr md5greet;
  estr md5verify;

  estr group;

  void sendGreet();
  void doHandshake();
  void doConnected();

  virtual void doSend();
  virtual void doRecv();
  virtual void doClose();
  virtual void doWait();
};


class edistcomp;
class edistcompQueue;

/*
class edistcompClient : public eworker
{
 public:
  edistcomp& server;
  estr recvbuffer;
  estr sendbuffer;
  evar result;

  etaskBase *rTask;
  evararray rArgs;

  estr id;
  estr hostname;
  unsigned int pid;
  estr group;

  bool isChoked;
  bool isBusy;

  edistcompClient();
  edistcompClient(edistcomp& server);
  ~edistcompClient();

  void choke();
  void unchoke();

  bool sendMsg(int code,const estr& msg);

  void call(const estr& cmd,const evararray& arr); 
  void interpret(const estr& cmd); 
  evar executeAtom(eatom_base *atom,estrhashof<evar>& env);

  virtual void execute(etaskBase& task,const efunc& func,const evararray& args);
  virtual void dispatch();

  void doProcess();

  virtual void doWait()=0;
  virtual void doSend()=0;
  virtual void doRecv()=0;
};

class edistcompClient_proc : public edistcompClient
{
 public:
  efile f;
  esystemCallback *fCallback;

  edistcompClient_proc();
  edistcompClient_proc(edistcomp& server,const efile& f);
//  ~edistcompClient();

  void init();

  void doWait();
  void doSend();
  void doRecv();
};

class edistcompClient_unix : public edistcompClient,public enamedsocket
{
 public:
  edistcompClient_unix();
  edistcompClient_unix(edistcomp& server,enamedsocket& s);

  void doWait();
  void doSend();
  void doRecv();
};

class edistcompClient_tcp : public edistcompClient,public esocket
{
 public:
  edistcompClient_tcp();
  edistcompClient_tcp(edistcomp& server,esocket& s);

  int handshake;
  estr md5greet;
  estr md5verify;

  void doHandshake();
  void doWait();
  void doSend();
  void doRecv();
};
*/

/*
class edistcompQueue;

class edctask
{
 public:
  bool isDone;
  int total;
  int count;
  int done;
  evararray args;
  efunc     func;
  evararray results;

  edctask();
  edctask(efunc& func);
  edctask(efunc& func,evararray& args);

  efunc& getFunc();
  evararray& getArgs();
  evar getResult();

  bool setRunning();
  bool doError();
  bool doFinished(evar& res);
  bool isFinished();
};

class edistcompGroup;

class edistcompQueue
{
 public:
  deque<edctask*> queued;
  list<edctask*>  running;
  earray<edctask>  finished;

  void addTask(const edctask &task);
  void runTask(edistcompClient& client);
  void doFinished(edistcompClient& client,edctask *ptask,evar& result);
  void doError(edctask *ptask);
  inline int queueCount(){ return(queued.size()); }
  inline int runCount(){ return(running.size()); }
  inline int finishedCount(){ return(finished.size()); }
};

class edistcompGroup
{
 public:
  edistcomp& dcserver;
  estrarray clients;
  edistcompQueue queue;

  edistcompGroup(edistcomp *dcserver);
  void startQueue();
  void addTask(const edctask& task);
};
*/

class evarRemote;

inline void evarRemoteFree(evarRemote& rvar);

class evarRemote // : public evarBase
{
 public:
  edcnode *dcnode;
  estr exechost;
  int execpid;
  long rvid;
  evarRemote(const evarRemote& o): dcnode(o.dcnode),exechost(o.exechost),execpid(o.execpid),rvid(o.rvid) {}
  evarRemote(): dcnode(0x00),execpid(0),rvid(0) {}
  evarRemote(edcnode *_dcnode,const estr& _exechost,int _execpid,long _rvid): dcnode(_dcnode),exechost(_exechost),execpid(_execpid),rvid(_rvid) {}
  virtual ~evarRemote() { evarRemoteFree(*this); }

  void serial(estr& s) const;
  void serial(const efile& f) const;

  long unserial(const estr& s,long i);
  bool unserial(const efile& f);
/*
  inline const type_info& getTypeid() { return(typeid(evarRemote)); }
  inline const type_info& getPTypeid() { return(typeid(evarRemote*)); }
  inline bool isTypeid(const type_info& tid) { return(tid==typeid(evarRemote) || tid==typeid(evarRemote*) || tid==typeid(evarRemote&) || tid==typeid(const evarRemote&)); }
  inline const char *getClass() { return(typeid(evarRemote).name()); }
  inline evarBase *copy() { return(new evarRemote(*this)); }
*/
};


inline void evarRemoteFree(evarRemote& rvar)
{
  if (rvar.dcnode==0x00) return;
  rvar.dcnode->remoteFree(rvar);
}

inline evar evarRemoteMethod(evarBase* v,const estr& mname,const evararray& args)
{
  evarRemote &rvar(*dynamic_cast<evarType<evarRemote>*>(v)->object);
  return(rvar.dcnode->remoteMethod(rvar,mname,args));
  // remote method call
}

inline evar evarRemoteProperty(evarBase* v,const estr& pname)
{
  evarRemote &rvar(*dynamic_cast<evarType<evarRemote>*>(v)->object);
  return(rvar.dcnode->remoteProperty(rvar,pname));
}

inline evar evarRemoteValue(evarBase* v)
{
  evarRemote &rvar(*dynamic_cast<evarType<evarRemote>*>(v)->object);
  return(rvar.dcnode->remoteValue(rvar));
}

inline void evarRemoteValue(evar& v)
{
  ldieif(v.var==0x00,"null evarbase");
  evarRemote *rvar(dynamic_cast<evarType<evarRemote>*>(v.var)->object);
  v.set(rvar->dcnode->remoteValue(*rvar));
}

inline estr evarRemoteExecuteHost(evarBase* v)
{
  evarRemote &rvar(*dynamic_cast<evarType<evarRemote>*>(v)->object);
  return(rvar.exechost);
}



class edistcomp
{
 public:
  bool showResult;
//  earray<efunc> callbacks;

  estrhash aliases;
  estrhashof<edcnode> clients;
  estrhashof<etaskQueue> groups;
  estrhashof<edcnode*> hosts;

//  efunc onAllReady;

  edistcomp();

#ifndef _WIN32
  enamedsocket nmserver;
#endif
  esocket      tcpserver;

  bool startProcClient(const estr& host,const estr& sfile=estr());
  void startTCPClient(esocket& s);
  void startTCPClient(esocket& s,const efunc& f);

#if !defined(_WIN32)
  void startUnixClient(enamedsocket& s);
  void startUnixClient(enamedsocket& s,const efunc& f);
#endif

//  void doAllReady();
//  void doReady(edistcompClient& dclient);
  virtual edcnode& getClient(int i);

/*
  void doHandleOutput(edistcompClient& client,const estr& data);
  void doHandleResult(edistcompClient& client,const estr& data);
  void doHandleHostname(edistcompClient& client,const estr& data);
*/

  virtual int nodeCount();

  edcnode* getClientByName(const estr& cname);

//  evar var(const estr& exechost,const estr& str);
  earray<estr> autocomplete(const estr& exechost,const estr& str);
  evar interpret(const estr& exechost,const estr& str);
  evar executeAtom(const estr& exechost,eatom_base *atom,estrhashof<evar>& env);
  evar executeCodeAtom(const estr& exechost,ecodeAtom *catom,estrhashof<evar>& env);


/*
  void objectCreate(evarRemote& v,const estr& type);
  void objectDelete(evarRemote& v);
  evar objectMethod(const evarRemote& v,const estr& mname,const evararray& args);
  evar objectGetProperty(const evarRemote& v,const estr& name);
  evar objectAssignProperty(const evarRemote& v,const estr& name,const evar& arg);
*/

  void fileClose(const edcfile& f);
  bool fileOpen(edcfile& f);
  void fileFlush(const edcfile& f);
  bool fileEOF(const edcfile& f);
  int fileWrite(const edcfile& f,const estr& str);
  int fileRead(const edcfile& f,estr& str,long int len);
  int fileReadLn(const edcfile& f,estr& str);
};

edistcomp& getDistComp();

void startDistCompLocalNode();
void startDistCompRemoteNode(const estr& host,const estr& group="");

#ifndef _WIN32
void startDistCompUnixNode(const estr& socketname);
#endif

#endif