/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

/*******************************************************************************
 * $Id: we_readthread.cpp 4609 2013-04-19 15:32:02Z chao $
 *
 *******************************************************************************/

#include <unistd.h>

#include "messagequeue.h"
#include "bytestream.h"
using namespace messageqcpp;

#include "threadpool.h"
using namespace threadpool;

#include "we_dataloader.h"
#include "we_readthread.h"
#include "we_messages.h"
#include "we_message_handlers.h"
#include "we_ddlcommandproc.h"
#include "we_dmlcommandproc.h"
#include "../redistribute/we_redistribute.h"
#include "we_config.h"
#include "stopwatch.h"
using namespace logging;
using namespace WriteEngine;
// StopWatch timer;
namespace WriteEngine
{
ReadThread::ReadThread(const IOSocket& ios) : fIos(ios)
{
}

ReadThread::~ReadThread()
{
}

void ReadThread::operator()()
{
  // We should never come here
}

//-----------------------------------------------------------------------------
// ctor
DmlReadThread::DmlReadThread(const messageqcpp::IOSocket& ios, ByteStream& Ibs)
 : ReadThread(ios), fWeDMLprocessor(new WE_DMLCommandProc), fWeDDLprocessor(new WE_DDLCommandProc)
{
  fIbs = Ibs;
}
// dtor
DmlReadThread::~DmlReadThread()
{
  // cout << "in DmlReadThread destructor" << endl;
}

void DmlReadThread::operator()()
{
  // DCH Since fIbs is a class member, there's no reason to make a copy here.
  // Why waste the CPU? Note that Splitter thread doesn't make a copy.
  // The only reason I can think of to make such a copy is to guarantee a
  // strong exception, but that doesn't appear to be in play here.
  ByteStream ibs = fIbs;
  ByteStream obs;
  ByteStream::byte msgId;
  ByteStream::octbyte uniqueID;
  ByteStream::quadbyte PMId;
  ByteStream::byte rc = 0;
  std::string errMsg;
  // cout << "DmlReadThread created ..." << endl;
  // queryStats.blocksChanged for delete/update
  uint64_t blocksChanged = 0;

  while (ibs.length() > 0)
  {
    try
    {
      errMsg.clear();

      // do work here...
      ibs >> msgId;

      if (msgId != WE_SVR_CLOSE_CONNECTION)
        ibs >> uniqueID;

      // cout << "DmlReadThread " << pthread_self () << " received message id " << (uint32_t)msgId << " and
      // bytestream length " << ibs.length() << endl;
      switch (msgId)
      {
        case WE_SVR_SINGLE_INSERT:
        {
          rc = fWeDMLprocessor->processSingleInsert(ibs, errMsg);
          break;
        }

        case WE_SVR_COMMIT_VERSION:
        {
          rc = fWeDMLprocessor->commitVersion(ibs, errMsg);
          break;
        }

        case WE_SVR_ROLLBACK_BLOCKS:
        {
          rc = fWeDMLprocessor->rollbackBlocks(ibs, errMsg);
          break;
        }

        case WE_SVR_ROLLBACK_VERSION:
        {
          rc = fWeDMLprocessor->rollbackVersion(ibs, errMsg);
          break;
        }

        case WE_SVR_COMMIT_BATCH_AUTO_ON:
        {
          rc = fWeDMLprocessor->commitBatchAutoOn(ibs, errMsg);
          break;
        }

        case WE_SVR_ROLLBACK_BATCH_AUTO_ON:
        {
          rc = fWeDMLprocessor->rollbackBatchAutoOn(ibs, errMsg);
          break;
        }

        case WE_SVR_COMMIT_BATCH_AUTO_OFF:
        {
          rc = fWeDMLprocessor->commitBatchAutoOn(ibs, errMsg);
          break;
        }

        case WE_SVR_ROLLBACK_BATCH_AUTO_OFF:
        {
          rc = fWeDMLprocessor->rollbackBatchAutoOff(ibs, errMsg);
          break;
        }

        case WE_SVR_BATCH_INSERT:
        {
          rc = fWeDMLprocessor->processBatchInsert(ibs, errMsg, PMId);
          break;
        }

        case WE_SVR_BATCH_INSERT_BINARY:
        {
          rc = fWeDMLprocessor->processBatchInsertBinary(ibs, errMsg, PMId);
          break;
        }

        case WE_SVR_GET_WRITTEN_LBIDS:
        {
          rc = fWeDMLprocessor->getWrittenLbids(ibs, errMsg, PMId);
          break;
        }

        case WE_SVR_BATCH_INSERT_END:
        {
          rc = fWeDMLprocessor->processBatchInsertHwm(ibs, errMsg);
          // timer.finish();
          break;
        }

        case WE_SVR_UPDATE:
        {
          rc = fWeDMLprocessor->processUpdate(ibs, errMsg, PMId, blocksChanged);
          break;
        }

        case WE_SVR_FLUSH_FILES:
        {
          rc = fWeDMLprocessor->processFlushFiles(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE:
        {
          rc = fWeDMLprocessor->processDelete(ibs, errMsg, PMId, blocksChanged);
          break;
        }

        case WE_SVR_BATCH_AUTOON_REMOVE_META:
        {
          rc = fWeDMLprocessor->processRemoveMeta(ibs, errMsg);
          break;
        }

        case WE_SVR_DML_BULKROLLBACK:
        {
          rc = fWeDMLprocessor->processBulkRollback(ibs, errMsg);
          break;
        }

        case WE_SVR_DML_BULKROLLBACK_CLEANUP:
        {
          rc = fWeDMLprocessor->processBulkRollbackCleanup(ibs, errMsg);
          break;
        }

        case WE_UPDATE_NEXTVAL:
        {
          rc = fWeDMLprocessor->updateSyscolumnNextval(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_SYSTABLE:
        {
          rc = fWeDDLprocessor->writeSystable(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_SYSCOLUMN:
        {
          rc = fWeDDLprocessor->writeSyscolumn(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_CREATE_SYSCOLUMN:
        {
          rc = fWeDDLprocessor->writeCreateSyscolumn(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_CREATETABLEFILES:
        {
          rc = fWeDDLprocessor->createtablefiles(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE_SYSCOLUMN:
        {
          rc = fWeDDLprocessor->deleteSyscolumn(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE_SYSCOLUMN_ROW:
        {
          rc = fWeDDLprocessor->deleteSyscolumnRow(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE_SYSTABLE:
        {
          rc = fWeDDLprocessor->deleteSystable(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE_SYSTABLES:
        {
          rc = fWeDDLprocessor->deleteSystables(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_DROPFILES:
        {
          rc = fWeDDLprocessor->dropFiles(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_AUTO:
        {
          rc = fWeDDLprocessor->updateSyscolumnAuto(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_NEXTVAL:
        {
          rc = fWeDDLprocessor->updateSyscolumnNextvalCol(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_AUTOVAL:
        {
          rc = fWeDDLprocessor->updateSyscolumnNextval(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_DEFAULTVAL:
        {
          rc = fWeDDLprocessor->updateSyscolumnSetDefault(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_TABLENAME:
        {
          rc = fWeDDLprocessor->updateSyscolumnTablename(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_RENAMECOLUMN:
        {
          rc = fWeDDLprocessor->updateSyscolumnRenameColumn(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSCOLUMN_COLPOS:
        {
          rc = fWeDDLprocessor->updateSyscolumnColumnposCol(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSTABLE_AUTO:
        {
          rc = fWeDDLprocessor->updateSystableAuto(ibs, errMsg);
          break;
        }

        case WE_SVR_UPDATE_SYSTABLE_TABLENAME:
        {
          rc = fWeDDLprocessor->updateSystableTablename(ibs, errMsg);
          break;
        }

        case WE_SVR_FILL_COLUMN:
        {
          rc = fWeDDLprocessor->fillNewColumn(ibs, errMsg);
          break;
        }

        case WE_SVR_DROP_PARTITIONS:
        {
          rc = fWeDDLprocessor->dropPartitions(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_TRUNCATE:
        {
          rc = fWeDDLprocessor->writeTruncateLog(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_DROPPARTITION:
        {
          rc = fWeDDLprocessor->writeDropPartitionLog(ibs, errMsg);
          break;
        }

        case WE_SVR_WRITE_DROPTABLE:
        {
          rc = fWeDDLprocessor->writeDropTableLog(ibs, errMsg);
          break;
        }

        case WE_SVR_DELETE_DDLLOG:
        {
          rc = fWeDDLprocessor->deleteDDLLog(ibs, errMsg);
          break;
        }

        case WE_SVR_FETCH_DDL_LOGS:
        {
          rc = fWeDDLprocessor->fetchDDLLog(ibs, errMsg);
          break;
        }

        case WE_SVR_PURGEFD:
        {
          rc = fWeDMLprocessor->processPurgeFDCache(ibs, errMsg);
          break;
        }

        case WE_END_TRANSACTION:
        {
          rc = fWeDMLprocessor->processEndTransaction(ibs, errMsg);
          break;
        }

        case WE_SRV_FIX_ROWS:
        {
          rc = fWeDMLprocessor->processFixRows(ibs, errMsg, PMId);
          break;
        }

        case WE_SVR_CLOSE_CONNECTION:
        {
          break;
        }

        default: break;
      }
    }
    catch (std::exception& ex)
    {
      logging::LoggingID logid(19, 0, 0);
      logging::Message::Args args;
      logging::Message msg(1);
      args.add("we_readthread caught exception ");
      args.add(ex.what());
      msg.format(args);
      logging::Logger logger(logid.fSubsysID);
      logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
      rc = 1;
      errMsg = msg.msg();
    }
    catch (...)
    {
      logging::LoggingID logid(19, 0, 0);
      logging::Message::Args args;
      logging::Message msg(1);
      args.add("we_readthread caught ... exception ");
      msg.format(args);
      logging::Logger logger(logid.fSubsysID);
      logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
      rc = 1;
      errMsg = msg.msg();
    }

    if (msgId != WE_SVR_CLOSE_CONNECTION)
    {
      // send response
      obs.restart();
      obs << uniqueID;
      obs << rc;
      obs << errMsg;
    }

    if ((msgId == WE_SVR_COMMIT_BATCH_AUTO_ON) || (msgId == WE_SVR_BATCH_INSERT_END) ||
        (msgId == WE_SVR_FETCH_DDL_LOGS) || (msgId == WE_SVR_GET_WRITTEN_LBIDS))
    {
      obs += ibs;
      // cout << " sending back hwm info with ibs length " << endl;
    }
    else if ((msgId == WE_SVR_BATCH_INSERT) || (msgId == WE_SVR_UPDATE) || (msgId == WE_SVR_DELETE))
    {
      obs << PMId;
    }
    else if ((msgId == WE_SVR_DML_BULKROLLBACK) || (msgId == WE_SVR_DML_BULKROLLBACK_CLEANUP))
    {
      obs << Config::getLocalModuleID();
    }

    if (msgId == WE_SVR_UPDATE || msgId == WE_SVR_DELETE)
      obs << blocksChanged;  // send stats back to DMLProc

    blocksChanged = 0;  // reset

    if (msgId == WE_SVR_CLOSE_CONNECTION)
    {
      // cout << "received request. closing connection ..." << endl;
      break;
    }
    else
    {
      try
      {
        fIos.write(obs);
        // cout << "dmlthread sent back response for msgid " << (uint32_t)msgId << " with uniqueID:rc= "
        //<< (uint32_t)uniqueID<<":"<< (uint32_t)rc<<" and error message is " << errMsg <<endl;
        // get next message
        ibs = fIos.read();
      }
      catch (...)
      {
        break;
      }
    }
  }

  // cout << "closing connection for thread " << pthread_self () << endl;
  fIos.close();
}

//-----------------------------------------------------------------------------
// ctor
SplitterReadThread::SplitterReadThread(const messageqcpp::IOSocket& ios, ByteStream& Ibs)
 : ReadThread(ios), fWeDataLoader(*this)
{
  fIbs = Ibs;
}

//-----------------------------------------------------------------------------
// copy ctor
SplitterReadThread::SplitterReadThread(const SplitterReadThread& rhs)
 : ReadThread(rhs.fIos), fWeDataLoader(*this)
{
  fIbs = rhs.fIbs;
}
//-----------------------------------------------------------------------------
// dtor
SplitterReadThread::~SplitterReadThread()
{
}
//-----------------------------------------------------------------------------
/**
 * @brief Thread Function which process incoming messages
 *
 */
void SplitterReadThread::operator()()
{
  ByteStream::byte msgId;

  while (fIbs.length() > 0)
  {
    fWeDataLoader.updateRxBytes(fIbs.length());

    // do work here...
    fIbs >> msgId;

    // cout << (int)msgId << endl;

    switch (msgId)
    {
      case WE_CLT_SRV_KEEPALIVE:
      {
        fWeDataLoader.onReceiveKeepAlive(fIbs);
        break;
      }

      case WE_CLT_SRV_DATA:
      {
        fWeDataLoader.onReceiveData(fIbs);
        break;
      }

      case WE_CLT_SRV_EOD:
      {
        fWeDataLoader.onReceiveEod(fIbs);
        break;
      }

      case WE_CLT_SRV_MODE:
      {
        fWeDataLoader.onReceiveMode(fIbs);
        break;
      }

      case WE_CLT_SRV_IMPFILENAME:
      {
        fWeDataLoader.onReceiveImportFileName(fIbs);
        break;
      }

      case WE_CLT_SRV_CMDLINEARGS:
      {
        fWeDataLoader.onReceiveCmdLineArgs(fIbs);
        break;
      }

      case WE_CLT_SRV_CMD:
      {
        fWeDataLoader.onReceiveCmd(fIbs);  // fig out share_ptr on BS& is better
        break;
      }

      case WE_CLT_SRV_ACK:
      {
        fWeDataLoader.onReceiveAck(fIbs);
        break;
      }

      case WE_CLT_SRV_NAK:
      {
        fWeDataLoader.onReceiveNak(fIbs);
        break;
      }

      case WE_CLT_SRV_PM_ERROR:
      {
        fWeDataLoader.onReceiveError(fIbs);
        break;
      }

      case WE_CLT_SRV_STARTCPI:
      {
        fWeDataLoader.onReceiveStartCpimport();
        break;
      }

      case WE_CLT_SRV_BRMRPT:
      {
        fWeDataLoader.onReceiveBrmRptFileName(fIbs);
        break;
      }

      case WE_CLT_SRV_CLEANUP:
      {
        fWeDataLoader.onReceiveCleanup(fIbs);
        break;
      }

      case WE_CLT_SRV_ROLLBACK:
      {
        fWeDataLoader.onReceiveRollback(fIbs);
        break;
      }

      case WE_CLT_SRV_JOBID:
      {
        fWeDataLoader.onReceiveJobId(fIbs);
        break;
      }

      case WE_CLT_SRV_JOBDATA:
      {
        fWeDataLoader.onReceiveJobData(fIbs);
        break;
      }

      case WE_CLT_SRV_ERRLOG:
      {
        fWeDataLoader.onReceiveErrFileRqst(fIbs);
        break;
      }

      case WE_CLT_SRV_BADLOG:
      {
        fWeDataLoader.onReceiveBadFileRqst(fIbs);
        break;
      }

      default: break;
    }

    fIbs.restart();

    try
    {
      // get next message
      fIbs = fIos.read();
    }
    catch (...)
    {
      fIbs.restart();  // setting length=0, get out of loop
      std::cout << "Broken Pipe" << std::endl;

      logging::LoggingID logid(19, 0, 0);
      logging::Message::Args args;
      logging::Message msg(1);
      args.add("SplitterReadThread::operator: Broken Pipe ");
      msg.format(args);
      logging::Logger logger(logid.fSubsysID);
      logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
    }
  }

  fIos.close();
}

//------------------------------------------------------------------------------
// ClearTableLockReadThread constructor.
//------------------------------------------------------------------------------
ClearTableLockReadThread::ClearTableLockReadThread(const messageqcpp::IOSocket& ios, ByteStream& Ibs)
 : ReadThread(ios), fWeClearTableLockCmd(new WE_ClearTableLockCmd("ClearTableLockTool"))
{
  fIbs = Ibs;
}

//------------------------------------------------------------------------------
// ClearTableLockReadThread destructor.
//------------------------------------------------------------------------------
ClearTableLockReadThread::~ClearTableLockReadThread()
{
}

//------------------------------------------------------------------------------
// Thread entry point to ClearTableLockReadThread object used to receive msgs
// from a cleartablelock tool client.
//------------------------------------------------------------------------------
void ClearTableLockReadThread::operator()()
{
  ByteStream::byte msgId;
  ByteStream obs;
  ByteStream::byte rc = 0;
  std::string errMsg;

  // Read msgid from ByteStream and forward to applicable processing function
  while (fIbs.length() > 0)
  {
    fIbs >> msgId;

    switch (msgId)
    {
      case WE_CLT_SRV_CLEAR_TABLE_LOCK:
      {
        rc = fWeClearTableLockCmd->processRollback(fIbs, errMsg);
        break;
      }

      case WE_CLT_SRV_CLEAR_TABLE_LOCK_CLEANUP:
      {
        rc = fWeClearTableLockCmd->processCleanup(fIbs, errMsg);
        break;
      }

      default:
      {
        break;
      }
    }

    // Send response
    obs.restart();
    obs << rc;
    obs << errMsg;

    try
    {
      fIos.write(obs);

      // Get next message
      fIbs = fIos.read();
    }
    catch (...)
    {
      logging::LoggingID logid(19, 0, 0);
      logging::Message::Args args;
      logging::Message msg(1);
      args.add("ClearTableLockReadThread::operator: Broken Pipe ");
      msg.format(args);
      logging::Logger logger(logid.fSubsysID);
      logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
      break;
    }
  }

  fIos.close();
}

//------------------------------------------------------------------------------
// RedistributeReadThread constructor.
//------------------------------------------------------------------------------
RedistributeReadThread::RedistributeReadThread(const messageqcpp::IOSocket& ios, ByteStream& Ibs)
 : ReadThread(ios)
{
  fIbs = Ibs;
}

//------------------------------------------------------------------------------
// RedistributeReadThread destructor.
//------------------------------------------------------------------------------
RedistributeReadThread::~RedistributeReadThread()
{
}

//------------------------------------------------------------------------------
// Thread entry point to RedistributeReadThread object used to receive msgs
// from a cleartablelock tool client.
//------------------------------------------------------------------------------
void RedistributeReadThread::operator()()
{
  try
  {
    redistribute::Redistribute::handleRedistributeMessage(fIbs, fIos);
  }
  catch (...)
  {
    logging::LoggingID logid(19, 0, 0);
    logging::Message::Args args;
    logging::Message msg(1);
    args.add("RedistributeReadThread::operator exception handled ");
    msg.format(args);
    logging::Logger logger(logid.fSubsysID);
    logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
  }

  fIos.close();
}
//------------------------------------------------------------------------------
// GetFileSizeThread constructor.
//------------------------------------------------------------------------------
GetFileSizeThread::GetFileSizeThread(const messageqcpp::IOSocket& ios, ByteStream& Ibs, BRM::DBRM& dbrm)
 : ReadThread(ios), fWeGetFileSizes(new WE_GetFileSizes())
{
  fIbs = Ibs;
  key = dbrm.getUnique32();
}

//------------------------------------------------------------------------------
// GetFileSizeThread destructor.
//------------------------------------------------------------------------------
GetFileSizeThread::~GetFileSizeThread()
{
}

//------------------------------------------------------------------------------
// Thread entry point to GetFileSizeThread object used to receive msgs
//------------------------------------------------------------------------------
void GetFileSizeThread::operator()()
{
  ByteStream::byte msgId;
  ByteStream obs;
  ByteStream::byte rc = 0;
  std::string errMsg;

  // Read msgid from ByteStream and forward to applicable processing function
  while (fIbs.length() > 0)
  {
    fIbs >> msgId;

    switch (msgId)
    {
      case WE_SVR_GET_FILESIZES:
      {
        rc = fWeGetFileSizes->processTable(fIbs, errMsg, key);
        break;
      }

      case WE_SVR_GET_FILESIZE:
      {
        rc = fWeGetFileSizes->processFileName(fIbs, errMsg, key);
        break;
      }

      default:
      {
        break;
      }
    }

    // Send response
    obs.restart();
    obs << rc;
    obs << errMsg;
    obs += fIbs;

    try
    {
      fIos.write(obs);

      // Get next message
      fIbs = fIos.read();
    }
    catch (...)
    {
      logging::LoggingID logid(19, 0, 0);
      logging::Message::Args args;
      logging::Message msg(1);
      args.add("GetFileSizeThread::operator: Broken Pipe ");
      msg.format(args);
      logging::Logger logger(logid.fSubsysID);
      logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
      break;
    }
  }

  fIos.close();
}

//-----------------------------------------------------------------------------

void ReadThreadFactory::CreateReadThread(ThreadPool& Tp, IOSocket& Ios, BRM::DBRM& dbrm)
{
  struct timespec rm_ts;
  int sleepTime = 20000;  // wait for 20 seconds
  rm_ts.tv_sec = sleepTime / 1000;
  rm_ts.tv_nsec = sleepTime % 1000 * 1000000;
  bool isTimeOut = false;

  ByteStream::byte msgId;
  ByteStream aBs;

  try
  {
    aBs = Ios.read(&rm_ts, &isTimeOut);
  }
  catch (std::exception& ex)
  {
    std::cout << "Handled : " << ex.what() << std::endl;
    logging::LoggingID logid(19, 0, 0);
    logging::Message::Args args;
    logging::Message msg(1);
    args.add("ReadThreadFactory::CreateReadThread: read() error");
    args.add(ex.what());
    msg.format(args);
    logging::Logger logger(logid.fSubsysID);
    logger.logMessage(logging::LOG_TYPE_ERROR, msg, logid);
  }

  if ((aBs.length() <= 0) || (isTimeOut))
  {
    Ios.close();
    return;
  }

  aBs.peek(msgId);

  switch (msgId)
  {
    case WE_SVR_DDL_KEEPALIVE:
    case WE_SVR_DML_KEEPALIVE:
    {
      DmlReadThread dmlReadThread(Ios, aBs);
      boost::thread t(dmlReadThread);
      // cout << "starting DML thread id " << t.get_id() << endl;
    }
    break;

    case WE_CLT_SRV_KEEPALIVE:
    case WE_CLT_SRV_MODE:
    case WE_CLT_SRV_DATA:
    case WE_CLT_SRV_CMD:
    case WE_CLT_SRV_ACK:
    case WE_CLT_SRV_NAK:
    case WE_CLT_SRV_PM_ERROR:
    case WE_CLT_SRV_CMDLINEARGS:
    {
      // SplitterReadThread aSpReadThread(Ios, aBs);
      // fOwner.attach(reinterpret_cast<Observer*>(&(aSpReadThread.fWeDataLoader)));
      // Tp.invoke(aSpReadThread);
      Tp.invoke(SplitterReadThread(Ios, aBs));
    }
    break;

    case WE_CLT_SRV_CLEAR_TABLE_LOCK:
    case WE_CLT_SRV_CLEAR_TABLE_LOCK_CLEANUP:
    {
      ClearTableLockReadThread clearTableLockThread(Ios, aBs);
      Tp.invoke(clearTableLockThread);
    }
    break;

    case WE_SVR_REDISTRIBUTE:
    {
      RedistributeReadThread RedistributeReadThread(Ios, aBs);
      Tp.invoke(RedistributeReadThread);
    }
    break;

    case WE_SVR_GET_FILESIZES:
    case WE_SVR_GET_FILESIZE:
    {
      GetFileSizeThread getFileSizeThread(Ios, aBs, dbrm);
      Tp.invoke(getFileSizeThread);
    }
    break;

    default:
    {
      Ios.close();  // don't know who is this
    }
    break;
  }
}

}  // namespace WriteEngine