#include <tcl.h>

#include "../data_framework/GenericTosData.h"
#include "../data_framework/GenericL1Data.h"
#include "../data_framework/StandardCandles.h"
#include "../../shared/ThreadClass.h"
#include "../../shared/ThreadMonitor.h"
#include "../../shared/CommandDispatcher.h"
#include "../../shared/Messages.h"
#include "../../shared/ReplyToClient.h"
#include "../../shared/SelectableRequestQueue.h"
#include "../../shared/LogFile.h"
#include "ReportAlertsThread.h"

#include "TclThread.h"

void safeCallTcl(Tcl_Interp *interp,
                 std::string script,
                 std::string notes,
                 SocketInfo *socket = NULL)
{
  int r =
    Tcl_Eval(interp, script.c_str());
  if (r)
  {
    TclList msg;
    msg<<"TCL error"
       <<script
       <<r
       <<Tcl_GetStringResult(interp);
    char const *errorInfo = Tcl_GetVar(interp, "errorInfo", TCL_GLOBAL_ONLY);
    if (errorInfo)
      msg<<errorInfo;
    else
      msg<<"NULL";
    msg<<notes;
    LogFile::primary().sendString(msg, socket);
  }
}

/////////////////////////////////////////////////////////////////////
// TclSupportDataNode
//
// This is not a traditional data node.  This is at the top of a
// heirarchy of data nodes.  This makes use of some of the data node
// mechanisms, like using broadcasts to move requests into the
// right thread.  This will report out using traditional queues
// (i.e. RequestListener / RequestQueue).
//
// Each Tcl thread object will create one of these before starting
// the thread.  This object will find and release other data nodes
// in the correct thread.
/////////////////////////////////////////////////////////////////////

class TclSupportDataNode : public DataNode
{
private:
  enum { StartL1BM, StopL1BM, StartCandlesBM, StopCandlesBM, SetListenerBM };
  class StringBM : public BroadcastMessage
  {
  public:
    StringBM(std::string const &channel) : BroadcastMessage(channel) {}
    std::string s;
  };
  class CandlesBM : public BroadcastMessage
  {
  public:
    CandlesBM(std::string const &channel) : BroadcastMessage(channel) {}
    std::string symbol;
    int minutes;
  };
  class ListenerBM : public BroadcastMessage
  {
  public:
    ListenerBM(std::string const &channel) : BroadcastMessage(channel) {}
    TclThread *listener;
  };

  class L1Listener : public DataNodeListener, ::NoCopy, ::NoAssign
  {
  private:
    enum { wTos, wL1 };
    TclSupportDataNode *const _owner;
    const std::string _symbol;
    GenericL1DataNode *_l1Data;
    DataNodeLink *_l1Link;
    GenericTosDataNode *_tosData;
    DataNodeLink *_tosLink;
    void pushInfo(bool noNewTrade);
    void onWakeup(int msgId);
  public:
    L1Listener(TclSupportDataNode *owner, std::string const &symbol);
    ~L1Listener();
  };

  std::string const _name;

  typedef std::pair< std::string, int > CandleKey;
  struct CandleData
  {
    DataNodeLink *link;
    StandardCandles *node;
    StandardCandles::Epoch epoch;
    CandleData() : link(NULL), node(NULL), epoch(0) { }
  };
  typedef std::map< CandleKey, CandleData > CandleList;
  CandleList _candleList;
  enum { wCandle };
  void onWakeup(int msgId);
  time_t _lastCandleUpdate;
  void newCandleData();

  std::map< std::string, L1Listener * > _l1Listeners;
  TclThread *_listener;
  void startL1Impl(std::string const &symbol);
  void stopL1Impl(std::string const &symbol);
  void startCandlesImpl(std::string const &symbol, int minutes);
  void stopCandlesImpl(std::string const &symbol, int minutes);
  void sendCandleChanges(TclList const &changes, std::string const &notes);
  static void checkForUpdate(TclList &output, CandleData &data,
                             std::string const &symbol, int minutes,
                             bool initial);
  void onBroadcast(BroadcastMessage &message, int msgId);
  std::string startL1Channel();
  std::string stopL1Channel();
  std::string startCandlesChannel();
  std::string stopCandlesChannel();
  std::string setListenerChannel();

  TclSupportDataNode(DataNodeArgument const &args);
  friend class DataNode;
  ~TclSupportDataNode();

public:
  // These are thread safe
  void startL1(std::string const &symbol);
  void stopL1(std::string const &symbol);
  void startCandles(std::string const &symbol, int minutes);
  void stopCandles(std::string const &symbol, int minutes);
  void setListener(TclThread *listener);
  void executeScript(std::string const &script,
                     std::string const &notes =
                     "TclSupportDataNode::executeScript()");

  static DataNodeLink *find(TclSupportDataNode *&node,
                            std::string const &name);
};

void TclSupportDataNode::startCandles(std::string const &symbol, int minutes)
{
  CandlesBM *m = new CandlesBM(startCandlesChannel());
  m->symbol = symbol;
  m->minutes = minutes;
  m->send(getManager());
}

void TclSupportDataNode::stopCandles(std::string const &symbol, int minutes)
{
  CandlesBM *m = new CandlesBM(stopCandlesChannel());
  m->symbol = symbol;
  m->minutes = minutes;
  m->send(getManager());
}

static TclList encodeCandle(StandardCandles::SingleCandle const &candle)
{
  TclList result;
  result<<"open"<<candle.open
        <<"high"<<candle.high
        <<"low"<<candle.low
        <<"close"<<candle.close
        <<"volume"<<candle.volume;
  return result;
}

void TclSupportDataNode::checkForUpdate(TclList &output,
                                        CandleData &data,
                                        std::string const &symbol,
                                        int minutes,
                                        bool initial)
{ // If there is a change, we need to add four things to the output list:
  // symbol, minutes, transition, data.  If no changes are required, we don't
  // add anything to the output list.
  if (initial || (data.node->getEpoch() != data.epoch))
  {
    data.epoch = data.node->getEpoch();
    output<<symbol<<minutes;
    if (initial
        || (data.node->getLastTransition() != StandardCandles::ltNewCandle))
    {
      if (initial)
        output<<"initial";
      else
        output<<"reset";
      TclList candles;
      StandardCandles::CandleArray const &history =
        data.node->getHistory();
      for (StandardCandles::CandleArray::const_iterator it =
             history.begin();
           it != history.end();
           it++)
      {
        candles<<encodeCandle(*it);
      }
      output<<candles;
    }
    else
    {
      output<<"one";
      output<<encodeCandle(*(data.node->getHistory().rbegin()));
    }
  }
}

void TclSupportDataNode::sendCandleChanges(TclList const &changes,
                                           std::string const &notes)
{
  if (changes.empty())
    return;
  TclList command;
  command<<"tiServer::new_data_candles"<<changes;
  _listener->executeScript(command, notes);
}

void TclSupportDataNode::startCandlesImpl(std::string const &symbol,
                                          int minutes)
{
  const CandleKey key(symbol, minutes);
  CandleData &data = _candleList[key];
  if (data.link)
  {
    TclList msg;
    msg<<__FILE__<<__LINE__<<__FUNCTION__
       <<"duplicate candle request"
       <<_name<<symbol<<minutes;
    LogFile::primary().sendString(msg);
  }
  else
  {
    data.link =
      StandardCandles::find(this, wCandle, data.node, symbol, minutes);
    TclList changes;
    checkForUpdate(changes, data, symbol, minutes, true);
    sendCandleChanges(changes, "TclSupportDataNode::startCandlesImpl()");
  }
}

void TclSupportDataNode::stopCandlesImpl(std::string const &symbol,
                                         int minutes)
{
  const CandleKey key(symbol, minutes);
  CandleData &data = _candleList[key];
  if (!data.link)
  {
    TclList msg;
    msg<<__FILE__<<__LINE__<<__FUNCTION__
       <<"unknown candle request"
       <<_name<<symbol<<minutes;
    LogFile::primary().sendString(msg);
  }
  else
  {
    data.link->release();
  }
  _candleList.erase(key);
}

void TclSupportDataNode::onWakeup(int msgId)
{
  switch (msgId)
  {
  case wCandle:
    // We handle candles a lot different than the way we handle L1 data.
    // We have different listeners for each symbol for the L1 data.  But
    // we update these all at once.  So we have just one listener covering
    // all stocks.
    newCandleData();
    break;
  }
}

void TclSupportDataNode::newCandleData()
{
  if (_lastCandleUpdate != getSubmitTime())
  {
    _lastCandleUpdate = getSubmitTime();
    TclList changes;
    for (CandleList::iterator it = _candleList.begin();
         it != _candleList.end();
         it++)
    {
      checkForUpdate(changes, it->second,
                     it->first.first, it->first.second,  false);
    }
    sendCandleChanges(changes, "TclSupportDataNode::newCandleData()");
  }
}

std::string TclSupportDataNode::startCandlesChannel()
{
  return getOwnerChannel() + ".startCandles";
}

std::string TclSupportDataNode::stopCandlesChannel()
{
  return getOwnerChannel() + ".stopCandles";
}

std::string TclSupportDataNode::startL1Channel()
{
  return getOwnerChannel() + ".startL1";
}

std::string TclSupportDataNode::stopL1Channel()
{
  return getOwnerChannel() + ".stopL1";
}

std::string TclSupportDataNode::setListenerChannel()
{
  return getOwnerChannel() + ".setListener";
}

void TclSupportDataNode::startL1Impl(std::string const &symbol)
{
  L1Listener *&listener = _l1Listeners[symbol];
  if (listener)
  {
    TclList msg;
    msg<<__FILE__<<__LINE__<<__FUNCTION__
       <<"duplicate symbol"
       <<_name<<symbol;
    LogFile::primary().sendString(msg);
  }
  else
  {
    listener = new L1Listener(this, symbol);
  }
}

void TclSupportDataNode::stopL1Impl(std::string const &symbol)
{
  L1Listener *&listener = _l1Listeners[symbol];
  if (!listener)
  {
    TclList msg;
    msg<<__FILE__<<__LINE__<<__FUNCTION__
       <<"unknown symbol"
       <<_name<<symbol;
    LogFile::primary().sendString(msg);
  }
  else
  {
    delete listener;
  }
  _l1Listeners.erase(symbol);
}

void TclSupportDataNode::onBroadcast(BroadcastMessage &message, int msgId)
{
  switch (msgId)
  {
  case StartL1BM:
    {
      StringBM &stringBM = dynamic_cast< StringBM & >(message);
      startL1Impl(stringBM.s);
      break;
    }
  case StopL1BM:
    {
      StringBM &stringBM = dynamic_cast< StringBM & >(message);
      stopL1Impl(stringBM.s);
      break;
    }
  case StartCandlesBM:
    {
      CandlesBM &candlesBM = dynamic_cast< CandlesBM & >(message);
      startCandlesImpl(candlesBM.symbol, candlesBM.minutes);
      break;
    }
  case StopCandlesBM:
    {
      CandlesBM &candlesBM = dynamic_cast< CandlesBM & >(message);
      stopCandlesImpl(candlesBM.symbol, candlesBM.minutes);
      break;
    }
  case SetListenerBM:
    {
      if (_listener)
      {
        TclList msg;
        msg<<__FILE__<<__LINE__<<__FUNCTION__
           <<"duplicate SetListenerBM"
           <<_name;
        LogFile::primary().sendString(msg);
        LogFile::primary().scheduleShutdown();
      }
      else
      {
        ListenerBM &listenerBM = dynamic_cast< ListenerBM & >(message);
        _listener = listenerBM.listener;
      }
      break;
    }
  }
}

TclSupportDataNode::TclSupportDataNode(DataNodeArgument const &args) :
  _name(args.getStringValue()),
  _lastCandleUpdate(0),
  _listener(NULL)
{
  registerForBroadcast(startL1Channel(), StartL1BM);
  registerForBroadcast(stopL1Channel(), StopL1BM);
  registerForBroadcast(startCandlesChannel(), StartCandlesBM);
  registerForBroadcast(stopCandlesChannel(), StopCandlesBM);
  registerForBroadcast(setListenerChannel(), SetListenerBM);
}

TclSupportDataNode::~TclSupportDataNode()
{
  for (std::map< std::string, L1Listener * >::iterator it =
         _l1Listeners.begin();
       it != _l1Listeners.end();
       it++)
    delete it->second;
}

void TclSupportDataNode::startL1(std::string const &symbol)
{
  StringBM *m = new StringBM(startL1Channel());
  m->s = symbol;
  m->send(getManager());
}

void TclSupportDataNode::stopL1(std::string const &symbol)
{
  StringBM *m = new StringBM(stopL1Channel());
  m->s = symbol;
  m->send(getManager());
}

void TclSupportDataNode::setListener(TclThread *listener)
{
  ListenerBM *m = new ListenerBM(setListenerChannel());
  m->listener = listener;
  m->send(getManager());
}

void TclSupportDataNode::executeScript(std::string const &symbol,
                                       std::string const &notes)
{
  if (!_listener)
  {
    TclList msg;
    msg<<__FILE__<<__LINE__<<__FUNCTION__
       <<"calling executeScript() before SetListenerBM"
       <<_name
       <<symbol<<notes;
    LogFile::primary().sendString(msg);
    LogFile::primary().scheduleShutdown();
  }
  else
  {
    _listener->executeScript(symbol, notes);
  }
}

DataNodeLink *TclSupportDataNode::find(TclSupportDataNode *&node,
                                       std::string const &name)
{ // This doesn't do traditional callbacks.  Instead people call
  // setListener() to get a response.  name can be anything.  The
  // point is that this should be unique.  No two callers should
  // find the same data node.
  return findHelper(NULL, 0, node, name);
}

/////////////////////////////////////////////////////////////////////
// TclSupportDataNode::L1Listener
//
// This owns the TOS and L1 nodes.  If the TclSupportDataNode owned
// these directly, then it would have trouble figuring out which
// symbol caused each event.
//
// Note the difference in names.  For historical reasons, our standard
// libraries seperate TOS and L1 data.  The TCL programmer will only
// see one combined record, which we call L1.
/////////////////////////////////////////////////////////////////////

template < class T >
inline void add(TclList &list, std::string const &name, T &value)
{
  list<<name<<value;
}

void TclSupportDataNode::L1Listener::pushInfo(bool noNewTrade)
{
  TclList data;
  if (_l1Data->getValid() && _tosData->getValid())
  {
    TosData const &last = _tosData->getLast();
    add(data, "lastPrice", last.price);
    add(data, "lastTime", last.time);
    add(data, "officialPrice", last.todaysClose);
    add(data, "lastSize", last.size);
    add(data, "lastExchange", last.exchange);
    add(data, "open", last.open);
    add(data, "high", last.high);
    add(data, "low", last.low);
    add(data, "formT", last.formT);
    add(data, "updatesLast", last.updatesLast);
    // For some reason this constant is required.  If I put that definition
    // directly into the line before, I get strange errors that don't make
    // sense.
    const bool newPrint = noNewTrade?false:last.newPrint;
    add(data, "newPrint", newPrint);
    //add(data, "newPrint", noNewTrade?false:last.newPrint);
    add(data, "volume", last.volume);
    L1Data const &current = _l1Data->getCurrent();
    add(data, "bidPrice", current.bidPrice);
    add(data, "bidSize", current.bidSize);
    add(data, "bidExchange", current.bidExchange);
    add(data, "askPrice", current.askPrice);
    add(data, "askSize", current.askSize);
    add(data, "askExchange", current.askExchange);
    //add(data, "prevClose", current.prevClose);
  }
  TclList command;
  command<<"tiServer::new_data_l1"
         <<_symbol
         <<data;
  _owner->executeScript(command, "TclSupportDataNode::L1Listener::pushInfo()");
}

void TclSupportDataNode::L1Listener::onWakeup(int msgId)
{
  pushInfo(msgId==wL1);
}

TclSupportDataNode::L1Listener::L1Listener(TclSupportDataNode *owner,
                                           std::string const &symbol) :
  _owner(owner),
  _symbol(symbol)
{
  _l1Link = GenericL1DataNode::find(this, wL1, _l1Data, symbol);
  _tosLink = GenericTosDataNode::find(this, wTos, _tosData, symbol);
  pushInfo(true);
}

TclSupportDataNode::L1Listener::~L1Listener()
{
  _l1Link->release();
  _tosLink->release();
}

/////////////////////////////////////////////////////////////////////
// TclThread
/////////////////////////////////////////////////////////////////////

TclThread::TclThread() :
  ThreadClass("TclThread " + pointerToString(this)),
  _incoming(getName()),
  _interp(NULL),
  _supportDataNode(NULL),
  _supportDataNodeLink(NULL)
{
  _supportDataNodeLink = TclSupportDataNode::find(_supportDataNode, getName());
  _supportDataNode->setListener(this);
  startThread();
}

TclThread::~TclThread()
{
  Request *r = new Request(NULL);
  r->callbackId = mtQuit;
  _incoming.newRequest(r);
  waitForThread();
  if (_supportDataNodeLink)
    _supportDataNodeLink->release();
}

void TclThread::executeScript(std::string const &script,
                              std::string const &notes)
{
  ScriptRequest *r = new ScriptRequest;
  r->script = script;
  r->notes = notes;
  r->callbackId = mtExecuteScript;
  _incoming.newRequest(r);
}

void TclThread::externalScript(std::string const &script,
                               SocketInfo *socket,
                               ExternalRequest::MessageId returnMsgId)
{
  ExternalScriptRequest *r = new ExternalScriptRequest(socket);
  r->script = script;
  r->returnMsgId = returnMsgId;
  r->callbackId = mtTclCommand;
  _incoming.newRequest(r);
}

static int requestL1Cmd(ClientData clientData,
                        Tcl_Interp *interp,
                        int objc,
                        Tcl_Obj *const objv[])
{
  if (objc != 2)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "symbol");
    return TCL_ERROR;
  }
  ((TclThread *)clientData)->requestL1Data(Tcl_GetString(objv[1]));
  return TCL_OK;
}

static int stopL1Cmd(ClientData clientData,
                     Tcl_Interp *interp,
                     int objc,
                     Tcl_Obj *const objv[])
{
  if (objc != 2)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "symbol");
    return TCL_ERROR;
  }
  ((TclThread *)clientData)->stopL1Data(Tcl_GetString(objv[1]));
  return TCL_OK;
}

static int requestCandlesCmd(ClientData clientData,
                             Tcl_Interp *interp,
                             int objc,
                             Tcl_Obj *const objv[])
{
  if (objc != 3)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "symbol minutes");
    return TCL_ERROR;
  }
  int minutes;
  if (Tcl_GetIntFromObj(interp, objv[2], &minutes) != TCL_OK)
    return TCL_ERROR;
  ((TclThread *)clientData)
    ->requestCandleData(Tcl_GetString(objv[1]), minutes);
  return TCL_OK;
}

static int stopCandlesCmd(ClientData clientData,
                          Tcl_Interp *interp,
                          int objc,
                          Tcl_Obj *const objv[])
{
  if (objc != 3)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "symbol minutes");
    return TCL_ERROR;
  }
  int minutes;
  if (Tcl_GetIntFromObj(interp, objv[2], &minutes) != TCL_OK)
    return TCL_ERROR;
  ((TclThread *)clientData)->stopCandleData(Tcl_GetString(objv[1]), minutes);
  return TCL_OK;
}

static std::string asString(Tcl_Obj *o)
{
  int length;
  char const *const p = Tcl_GetStringFromObj(o, &length);
  return std::string(p, length);
}

static int reportAlertCmd(ClientData clientData,
                          Tcl_Interp *interp,
                          int objc,
                          Tcl_Obj *const objv[])
{
  if (objc != 3)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "category info");
    return TCL_ERROR;
  }
  ThreadMonitor::find().increment("reportAlertCmd");
  ReportAlertsThread::getInstance()->reportAlert(asString(objv[1]),
                                                 asString(objv[2]));
  return TCL_OK;
}

static int addAlertCategoryCmd(ClientData clientData,
                               Tcl_Interp *interp,
                               int objc,
                               Tcl_Obj *const objv[])
{
  if (objc != 3)
  {
    Tcl_WrongNumArgs(interp, 1, objv, "category description");
    return TCL_ERROR;
  }
  ThreadMonitor::find().increment("addAlertCategoryCmd");
  ReportAlertsThread::getInstance()->addAlertCategory(asString(objv[1]),
                                                      asString(objv[2]));
  return TCL_OK;

}

void TclThread::threadFunction()
{
  ThreadMonitor &tm = ThreadMonitor::find();
  tm.setState("init");
  _interp = Tcl_CreateInterp();
  Tcl_Init(_interp);
  Tcl_CreateObjCommand(_interp, "tiServer::request_l1_from_cpp",
                       requestL1Cmd, this, NULL);
  Tcl_CreateObjCommand(_interp, "tiServer::stop_l1_from_cpp",
                       stopL1Cmd, this, NULL);
  Tcl_CreateObjCommand(_interp, "tiServer::request_candles_from_cpp",
                       requestCandlesCmd, this, NULL);
  Tcl_CreateObjCommand(_interp, "tiServer::stop_candles_from_cpp",
                       stopCandlesCmd, this, NULL);
  Tcl_CreateObjCommand(_interp, "tiServer::report_alert",
                       reportAlertCmd, NULL, NULL);
  Tcl_CreateObjCommand(_interp, "tiServer::add_alert_category",
                       addAlertCategoryCmd, NULL, NULL);
  safeCallTcl(_interp, "source server_code.tcl", "init");

  while (true)
  {
    _incoming.resetWaitHandle();
    while (Request *current = _incoming.getRequest())
    {
      tm.setState("working");
      switch (current->callbackId)
      {
        /*
          case mtDataNodeEvent:
          {
          DataNodeManager::EventQueueListener *request =
          dynamic_cast< DataNodeManager::EventQueueListener * >
          (current);
          request->onEventQueue();
          break;
          }
        */
      case mtTclCommand:
        {
          tm.setState("mtTclCommand");
          tm.increment("mtTclCommand");
          ExternalScriptRequest *request =
            dynamic_cast< ExternalScriptRequest * >(current);
          int r =
            Tcl_Eval(_interp, request->script.c_str());
          std::string msg = ntoa(r);
          msg += '\n';
          msg += Tcl_GetStringResult(_interp);
          addToOutputQueue(request->getSocketInfo(),
                           msg,
                           request->returnMsgId);
          break;
        }
      case mtExecuteScript:
        {
          tm.setState("mtExecuteScript");
          tm.increment("mtExecuteScript");
          ScriptRequest *request =
            dynamic_cast< ScriptRequest * >(current);
          safeCallTcl(_interp, request->script, request->notes);
          break;
        }
      case mtQuit:
        {
          // Free _interp;
          delete current;
          return;
        }
      }
      delete current;
    }
    // For simplicity, we read all messages from _incoming, then we read
    // all messages from the tcl event loop, then we check _incoming again.
    // There is no real priority between these two queues, and there is no
    // attempt to interleave them.
    tm.setState("Tcl_DoOneEvent");
    while (Tcl_DoOneEvent(TCL_DONT_WAIT))
    { // This should count the number of times we actually did an event, not the
      // number of times we tried.
      tm.increment("Tcl_DoOneEvent");
    }
    tm.increment("waitForRequest");
    tm.setState("waitForRequest");
    // To keep the event loop live in TCL, poll for events at least once every
    // 100 milliseconds.
    timeval pollInterval;
    pollInterval.tv_sec = 0;
    pollInterval.tv_usec = 100000;
    _incoming.waitForRequest(&pollInterval);
  }
}

void TclThread::requestL1Data(std::string const &symbol)
{
  _supportDataNode->startL1(symbol);
}

void TclThread::stopL1Data(std::string const &symbol)
{
  _supportDataNode->stopL1(symbol);
}

void TclThread::requestCandleData(std::string const &symbol, int minutes)
{
  _supportDataNode->startCandles(symbol, minutes);
}

void TclThread::stopCandleData(std::string const &symbol, int minutes)
{
  _supportDataNode->stopCandles(symbol, minutes);
}