<commit_before>/*
 * Copyright 2006-2008 The FLWOR Foundation.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "debugger/debugger_server.h"

#include <iostream>
#include <memory>
#include <fstream>

#include <zorba/zorba.h>
#include <zorba/util/path.h>

#include "debugger/utils.h"
#include "debugger/socket.h"
#include "debugger/message_factory.h"

#include "context/static_context.h"

#include "runtime/fncontext/FnContextImpl.h"
#include "runtime/core/item_iterator.h"
#include "runtime/api/runtimecb.h"
#include "runtime/api/plan_iterator_wrapper.h"
#include "runtime/util/iterator_impl.h"

#include "api/serialization/serializer.h"
#include "api/unmarshaller.h"

#include "zorbatypes/numconversions.h"

#include "common/shared_types.h"

#ifdef WIN32
#include <windows.h>
#define sleep(s) Sleep(s*1000)
#endif

namespace zorba{

ZORBA_THREAD_RETURN runtimeThread( void *aDebugger )
{
  ZorbaDebugger * lDebugger = (ZorbaDebugger *)aDebugger;
  //We wait for theRuntimeThread to be allocated
  while( lDebugger->theRuntimeThread == 0 )
  {
	sleep(1);
  }
  lDebugger->runQuery();
  return 0;
}

ZorbaDebugger::~ZorbaDebugger()
{
  delete theRequestServerSocket;
  delete theEventSocket;
}

void ZorbaDebugger::start( XQueryImpl *aQuery,
                           unsigned short aRequestPortno,
                           unsigned short aEventPortno)
{
  std::auto_ptr<TCPSocket> lSock;
  //Set the query
  theQuery = aQuery;
  //Run the server 
  theRequestServerSocket = new TCPServerSocket( aRequestPortno );
  theEventPortno = aEventPortno;
  
  //Wait for a client to connect
  lSock.reset(theRequestServerSocket->accept());
  assert( lSock.get() != 0 );
#ifndef NDEBUG
  std::clog << "[Server Thread] Client connected" << std::endl;
#endif
  //Try to connect to the event server 3 times
  for ( unsigned int i = 0; i < 3 && ! theEventSocket; i++ )
  {
    try
    {
      //Wait one second before trying to reconnect
      sleep(1);
      //Connect the client to the event server
      theEventSocket = new TCPSocket( "127.0.0.1", theEventPortno );
      break;
    } catch ( SocketException &e )  {
      if ( i < 2 ) continue;
      std::cerr << "[Server Thread] Couldn't connect to the debugger server event: " << std::endl;
      std::cerr << "[Server Thread] " <<  e.what() << std::endl;
      return;
    }
  }
#ifndef NDEBUG
  std::clog << "[Server Thread] Connected to the event server" << std::endl;
#endif

  //Perform handshake
  handshake( lSock.get() );
  
  //Until the query execution has ended
  while ( theStatus != QUERY_QUITED )
  {
    handleTcpClient( lSock.get() );
  }
#ifndef NDEBUG
  std::clog << "[Server Thread] server quited" << std::endl;
#endif
  delete theRuntimeThread;
#ifndef NDEBUG
  std::clog << "[Server Thread] runtime thread quited" << std::endl;
#endif 
}

void ZorbaDebugger::setStatus( ExecutionStatus Status,
                                   SuspensionCause aCause ){

  if ( theStatus == QUERY_SUSPENDED && Status == QUERY_RUNNING )
  {
    theStatus = QUERY_RESUMED;
  } else {
	  theStatus = Status;
  }

  switch ( theStatus )
  {
    case QUERY_RUNNING:
      startedEvent();
      break;
    case QUERY_SUSPENDED:
      suspendedEvent( theLocation, aCause );
      break;
    case QUERY_TERMINATED:
      terminatedEvent();
      break;
    case QUERY_RESUMED:
      theStatus = QUERY_RUNNING;
      resumedEvent();
      break;
  }
}

void ZorbaDebugger::sendEvent( AbstractCommandMessage * aMessage )
{
    Length length;
    ZorbaArrayAutoPointer<Byte> lMessage(aMessage->serialize(length));
    try
    {
#ifndef NDEBUG
      std::clog << "[Server Thread] send an event: ";
      switch ( aMessage->getCommand() )
      {
        case STARTED:
          std::cerr << "started" << std::endl;
          break;
        case TERMINATED:
          std::cerr << "terminated" << std::endl;
          break;
        case SUSPENDED:
          std::cerr << "suspended" << std::endl;
          break;
        case RESUMED:
          std::cerr << "resumed" << std::endl;
          break;
        case EVALUATED:
          std::cerr << "evaluated" << std::endl;
          break;
      }
#endif
      theEventSocket->send( lMessage.get(), length );
#ifndef NDEBUG
      std::clog << "[Server Thread] event sent" << std::endl;
#endif
    } catch( SocketException &e ) {
      std::cerr << e.what() << std::endl;
    }
}


void ZorbaDebugger::startedEvent()
{
    StartedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::suspendedEvent( QueryLoc &aLocation, SuspensionCause aCause )
{
    SuspendedEvent lMessage( aLocation, aCause );
    sendEvent( &lMessage );
}

void ZorbaDebugger::resumedEvent()
{
    ResumedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::terminatedEvent()
{
    TerminatedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::runQuery()
{
  setStatus( QUERY_RUNNING );
  try
  {
    Zorba_SerializerOptions lSerOptions;
    lSerOptions.omit_xml_declaration = ZORBA_OMIT_XML_DECLARATION_YES;
    theQuery->serialize( std::cout, lSerOptions );
  } catch ( StaticException& se ) {
    std::cerr << se << std::endl;
  } catch ( DynamicException& de ) {
    std::cerr << de << std::endl;
  } catch ( SystemException& se ) {
    std::cerr << se << std::endl;
  }
  setStatus( QUERY_TERMINATED );
}

bool ZorbaDebugger::hasToSuspend()
{
  if( theStatus == QUERY_SUSPENDED )
  {
    return true;
  }
  std::map<unsigned int, QueryLoc>::iterator it;
  //TODO: can be faster
  for ( it = theBreakpoints.begin(); it != theBreakpoints.end(); it++ )
  {
    std::string lCurrentFilename = theLocation.getFilename();
    std::string lBreakFilename = it->second.getFilename();
    int start = lCurrentFilename.length() - lBreakFilename.length();
    bool fileMatch = true;
    if( start >= 0){
      fileMatch = (lBreakFilename == lCurrentFilename.substr(start));
    }
    if (fileMatch && theLocation.getLineno() == it->second.getLineno() )
    {
      setStatus( QUERY_SUSPENDED, CAUSE_BREAKPOINT );
      return true;
    }
  }
 
  std::map<unsigned int, xqpString>::iterator lIter;
  for ( lIter = theWatchpoints.begin(); lIter != theWatchpoints.end(); lIter++ )
  {
    std::auto_ptr< CompilerCB > ccb(new CompilerCB(*thePlanState->theCompilerCB));
    std::auto_ptr< dynamic_context > dctx(new dynamic_context(thePlanState->dctx()));
      
    //TODO: catch exceptions
    std::auto_ptr<PlanWrapperHolder> eval_plan = compileEvalPlan(theLocation, ccb.get(), dctx.get(), lIter->second, *thePlanState);
    PlanWrapper* lIterator = eval_plan->get();
    try
    {
      store::Item_t lItem;
      std::map<xqpString, xqpString> lValuesAndTypes;

      error::ErrorManager lErrorManger;

      lIterator->next(lItem);
      
      if(lItem->getEBV()->getBooleanValue())
      {
        setStatus(QUERY_SUSPENDED, CAUSE_BREAKPOINT);
	      return true;
      }
    } catch ( error::ZorbaError& e) {
      //do nothing... 
    }
  }
  return false;
}

/**
 * Perform an handshake with the client.
 * Simply returns the received first 11 bytes.
 */
void ZorbaDebugger::handshake( TCPSocket * aSock )
{
  ZorbaArrayAutoPointer<Byte> msg(new Byte[11]);
  try
  {
    aSock->recv( msg.get(), 11 );
    aSock->send( msg.get(), 11 );
  } catch( SocketException &e ) {
    std::cerr << e.what() << std::endl;
  }
}

/**
 * TODO: check if the client has closed the connection
 */
void ZorbaDebugger::handleTcpClient( TCPSocket * aSock )
{
  ZorbaArrayAutoPointer<Byte> lByteMessage;
  std::auto_ptr<AbstractMessage> lMessage;
  std::auto_ptr<ReplyMessage> lReplyMessage;
  Length length;
  try
  {
    lMessage.reset(MessageFactory::buildMessage( aSock ));
    AbstractCommandMessage * lCommandMessage = dynamic_cast< AbstractCommandMessage * > ( lMessage.get() );
    if ( lCommandMessage )
    {
      //process the command message
      //TODO: flush the output somewhere else
      std::cout.flush();
      processMessage( lCommandMessage );
      //Send the reply message
      lReplyMessage.reset(lCommandMessage->getReplyMessage());
      lByteMessage.reset(lReplyMessage->serialize( length ));
      aSock->send( lByteMessage.get(), length );
    } else {
#ifndef NDEBUG
      std::cerr << "Received something wrong" << std::endl;
#endif
      //If something goes wrong, buildMessage() receive a Reply Message containing the error description
      //Send it back to the client right away
      lReplyMessage.reset(dynamic_cast<ReplyMessage *> ( lMessage.get() ));
      if( lReplyMessage.get() ){
        lByteMessage.reset(lReplyMessage->serialize( length ));
        aSock->send( lByteMessage.get(), length );
      } else {
        std::cerr << "Internal error occured. Couldn't send the error message." << std::endl;
      }
    }    
  } catch ( std::exception &e ) {
    std::cerr <<  e.what() << std::endl;
  }
}

void ZorbaDebugger::run()
{
  theRuntimeThread = new Thread( runtimeThread, this );
}

void ZorbaDebugger::suspend()
{
  setStatus( QUERY_SUSPENDED );
}

void ZorbaDebugger::resume()
{
  if ( theStatus == QUERY_SUSPENDED )
  {
    setStatus( QUERY_RUNNING );
    theRuntimeThread->resume();
  }
}

void ZorbaDebugger::terminate()
{
  setStatus( QUERY_TERMINATED );
}

void ZorbaDebugger::quit()
{
  setStatus( QUERY_QUITED );
}

void ZorbaDebugger::eval( xqpString anExpr )
{
  if ( theStatus != QUERY_IDLE && theStatus != QUERY_TERMINATED )
  {
    std::auto_ptr<EvaluatedEvent> lMsg;
    try {
      std::auto_ptr< CompilerCB > ccb(new CompilerCB(*thePlanState->theCompilerCB));
      std::auto_ptr< dynamic_context > dctx(new dynamic_context(thePlanState->dctx()));
      std::auto_ptr<PlanWrapperHolder> eval_plan = compileEvalPlan(theLocation, ccb.get(), dctx.get(), anExpr, *thePlanState);
      PlanWrapper* lIterator = eval_plan->get();
      assert(lIterator != 0);

      store::Item_t lItem;
      std::map<xqpString, xqpString> lValuesAndTypes;

      error::ErrorManager lErrorManger;
      serializer lSerializer(&lErrorManger);
      lSerializer.set_parameter("omit-xml-declaration", "yes");

      while (lIterator->next(lItem)) {
        std::ostringstream os;
        lSerializer.serialize(lItem, os);
        xqpString lValue = os.str();
        xqpString lType( lItem->getType()->getStringValue() );
        lValuesAndTypes.insert(std::pair<xqpString, xqpString>(lValue, lType));
      }
      lMsg.reset(new EvaluatedEvent(anExpr, lValuesAndTypes));
    } catch ( error::ZorbaError& e) {
      std::stringstream lOutputStream;
      xqpString lDescription = e.theDescription.replace("\\\"", "", "");
      lOutputStream << "Error: " << error::ZorbaError::toString(e.theErrorCode) << " " << lDescription;
      lMsg.reset(new EvaluatedEvent( anExpr, lOutputStream.str() ));
    }
    assert(lMsg.get() != 0);
    sendEvent(lMsg.get());
  }
}

std::auto_ptr<PlanWrapperHolder>
ZorbaDebugger::compileEvalPlan(const QueryLoc& loc, CompilerCB* ccb, dynamic_context* dctx, const xqpString& anExpr, PlanState& planState)
{
  std::auto_ptr<PlanWrapperHolder> eval_plan(new PlanWrapperHolder());
  
  checked_vector< std::string > var_keys;

  //set up eval state's ccb
  ccb->m_sctx_list.push_back( ccb->m_sctx = ccb->m_sctx->create_child_context() );
  ccb->m_debugger = 0;

  eval_plan->reset(
    new PlanWrapper (
      EvalIterator::compile ( ccb, anExpr, theVarnames, theVartypes ),
      ccb,
      dctx,
      planState.theStackDepth + 1 )
    );
  eval_plan->get()->checkDepth( loc );
  
  for (unsigned i = 0; i < theChildren.size () - 1; i++)
  {
    //reset the plan iterator for multiple execution
    theChildren[i+1]->reset(*thePlanState);
    store::Iterator_t lIter = new PlanIteratorWrapper(theChildren[i+1], *thePlanState);
    // TODO: is saving an open iterator efficient?
    // Then again if we close theChildren [1] here,
    // we won't be able to re-open it later via the PlanIteratorWrapper
    dctx->add_variable(dynamic_context::var_key(ccb->m_sctx->lookup_var(theVarnames[i])), lIter);
  }
  return std::auto_ptr<PlanWrapperHolder>(eval_plan.release());
}

void ZorbaDebugger::processMessage(AbstractCommandMessage * aMessage)
{
  switch( aMessage->getCommandSet() )
  {
    case EXECUTION:
    {
      switch( aMessage->getCommand() )
      {
        case RUN:
        {
#ifndef NDEBUG
          RunMessage * lMessage = dynamic_cast<RunMessage *> ( aMessage );
          assert( lMessage );
#endif
          run();
          break;
        }
        case SUSPEND:
        {
#ifndef NDEBUG
          SuspendMessage * lMessage = dynamic_cast<SuspendMessage *> ( aMessage );
          assert( lMessage );
#endif
          suspend();
          break;
        }
        case RESUME:
        {
#ifndef NDEBUG
          ResumeMessage * lMessage = dynamic_cast<ResumeMessage*> ( aMessage );
          assert( lMessage );
#endif
          resume();
          break;
        }
        case QUIT:
        {
#ifndef NDEGUB
          QuitMessage * lMessage = dynamic_cast<QuitMessage *>( aMessage );
          assert( lMessage );
#endif
          quit();
          break;
        }
        case TERMINATE:
        {
#ifndef NDEBUG
          TerminateMessage * lMessage = dynamic_cast<TerminateMessage *> ( aMessage );
          assert( lMessage );
#endif
          terminate();
          break;
        }
        case STEP:
        {
#ifndef NDEBUG
          StepMessage * lMessage = dynamic_cast<StepMessage *> ( aMessage );
          assert( lMessage );
          processMessage( lMessage );
#else
          processMessage( static_cast<StepMessage *> ( aMessage ));
#endif
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for execution command set .");
    }
      break;
    case BREAKPOINTS:
    {
      switch( aMessage->getCommand() )
      {
        case SET:
        {
          SetMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< SetMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage =  static_cast< SetMessage * > ( aMessage );
#endif
          std::map<unsigned int, QueryLoc> locations = lMessage->getLocations();
          std::map<unsigned int, QueryLoc>::iterator it;
          for ( it = locations.begin(); it != locations.end(); it++ )
          {
            theBreakpoints.insert( std::make_pair( it->first, it->second ) ); 
          }
          
          std::map<unsigned int, xqpString> exprs = lMessage->getExprs();
          std::map<unsigned int, xqpString>::iterator lIt;
          for ( lIt = exprs.begin(); lIt != exprs.end(); lIt++ )
          {
            theWatchpoints.insert( std::make_pair( lIt->first, lIt->second ) ); 
          }
          break;
        }
        case CLEAR:
        {
          ClearMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< ClearMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage =  static_cast< ClearMessage * > ( aMessage );
#endif
          std::vector<unsigned int>::iterator it;
          std::vector<unsigned int> ids = lMessage->getIds();
          for ( it = ids.begin(); it != ids.end(); it++ )
          {
            std::map<unsigned int, QueryLoc>::iterator
            lIter = theBreakpoints.find( *it );
            if ( lIter !=  theBreakpoints.end() )
            {
              theBreakpoints.erase( lIter );
            } else {
              std::map<unsigned int, xqpString>::iterator lIter2 =
              theWatchpoints.find( *it );
              if ( lIter2 != theWatchpoints.end() ) {
                theWatchpoints.erase( lIter2 );
              }
            }
          }
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for breakpoints command set.");
      }
    }
      break;
    case ENGINE_EVENT:
    case STATIC:
    case DYNAMIC:
    {
      switch ( aMessage->getCommand() )
      {
        case EVAL:
        {
          EvalMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< EvalMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage = static_cast< EvalMessage * > ( aMessage );
#endif
          eval( lMessage->getExpr() );
          break;
        }
        case VARIABLES:
        {
          VariableMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< VariableMessage * > ( aMessage );
#else
          lMessage = static_cast< VariableMessage * > ( aMessage );
#endif
          VariableReply * lReply = new VariableReply( lMessage->getId(), DEBUGGER_NO_ERROR );
          for( unsigned i = 0; i<theVarnames.size(); i++ )
          {
            xqpString lName(theVarnames.at(i)->getStringValue());
            xqpString lType;
            if( theVartypes.at(i) != 0 )
            {
              lType = theVartypes.at(i)->toString();
            } else {
              lType = "[XS_UNTYPED]";
            }
            checked_vector<global_binding>::iterator it;
            bool is_global = false;
            for ( it = theGlobals.begin(); it != theGlobals.end(); ++it )
            {
              if ( it->first->get_varname() == theVarnames.at(i) )
              {
                lReply->addGlobal( lName, lType );
                is_global = true;
                break;
              }
            }
            if ( ! is_global )
            {
              lReply->addLocal( lName, lType );
            }
          }
          lMessage->setReplyMessage( lReply );
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for dynamic command set.");
      }
    }
    break;
    default:
      throw InvalidCommandException("Internal Error. CommandSet not implemented.");
  }
}
};
}//end of namespace

<commit_msg>Flush debugger output.<commit_after>/*
 * Copyright 2006-2008 The FLWOR Foundation.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "debugger/debugger_server.h"

#include <iostream>
#include <memory>
#include <fstream>

#include <zorba/zorba.h>
#include <zorba/util/path.h>

#include "debugger/utils.h"
#include "debugger/socket.h"
#include "debugger/message_factory.h"

#include "context/static_context.h"

#include "runtime/fncontext/FnContextImpl.h"
#include "runtime/core/item_iterator.h"
#include "runtime/api/runtimecb.h"
#include "runtime/api/plan_iterator_wrapper.h"
#include "runtime/util/iterator_impl.h"

#include "api/serialization/serializer.h"
#include "api/unmarshaller.h"

#include "zorbatypes/numconversions.h"

#include "common/shared_types.h"

#ifdef WIN32
#include <windows.h>
#define sleep(s) Sleep(s*1000)
#endif

namespace zorba{

ZORBA_THREAD_RETURN runtimeThread( void *aDebugger )
{
  ZorbaDebugger * lDebugger = (ZorbaDebugger *)aDebugger;
  //We wait for theRuntimeThread to be allocated
  while( lDebugger->theRuntimeThread == 0 )
  {
	sleep(1);
  }
  lDebugger->runQuery();
  return 0;
}

ZorbaDebugger::~ZorbaDebugger()
{
  delete theRequestServerSocket;
  delete theEventSocket;
}

void ZorbaDebugger::start( XQueryImpl *aQuery,
                           unsigned short aRequestPortno,
                           unsigned short aEventPortno)
{
  std::auto_ptr<TCPSocket> lSock;
  //Set the query
  theQuery = aQuery;
  //Run the server 
  theRequestServerSocket = new TCPServerSocket( aRequestPortno );
  theEventPortno = aEventPortno;
  
  //Wait for a client to connect
  lSock.reset(theRequestServerSocket->accept());
  assert( lSock.get() != 0 );
#ifndef NDEBUG
  std::clog << "[Server Thread] Client connected" << std::endl;
#endif
  //Try to connect to the event server 3 times
  for ( unsigned int i = 0; i < 3 && ! theEventSocket; i++ )
  {
    try
    {
      //Wait one second before trying to reconnect
      sleep(1);
      //Connect the client to the event server
      theEventSocket = new TCPSocket( "127.0.0.1", theEventPortno );
      break;
    } catch ( SocketException &e )  {
      if ( i < 2 ) continue;
      std::cerr << "[Server Thread] Couldn't connect to the debugger server event: " << std::endl;
      std::cerr << "[Server Thread] " <<  e.what() << std::endl;
      return;
    }
  }
#ifndef NDEBUG
  std::clog << "[Server Thread] Connected to the event server" << std::endl;
#endif

  //Perform handshake
  handshake( lSock.get() );
  
  //Until the query execution has ended
  while ( theStatus != QUERY_QUITED )
  {
    handleTcpClient( lSock.get() );
  }
#ifndef NDEBUG
  std::clog << "[Server Thread] server quited" << std::endl;
#endif
  delete theRuntimeThread;
#ifndef NDEBUG
  std::clog << "[Server Thread] runtime thread quited" << std::endl;
#endif 
}

void ZorbaDebugger::setStatus( ExecutionStatus Status,
                                   SuspensionCause aCause ){

  if ( theStatus == QUERY_SUSPENDED && Status == QUERY_RUNNING )
  {
    theStatus = QUERY_RESUMED;
  } else {
	  theStatus = Status;
  }

  switch ( theStatus )
  {
    case QUERY_RUNNING:
      startedEvent();
      break;
    case QUERY_SUSPENDED:
      suspendedEvent( theLocation, aCause );
      break;
    case QUERY_TERMINATED:
      terminatedEvent();
      break;
    case QUERY_RESUMED:
      theStatus = QUERY_RUNNING;
      resumedEvent();
      break;
  }
}

void ZorbaDebugger::sendEvent( AbstractCommandMessage * aMessage )
{
    Length length;
    ZorbaArrayAutoPointer<Byte> lMessage(aMessage->serialize(length));
    try
    {
#ifndef NDEBUG
      std::clog << "[Server Thread] send an event: ";
      switch ( aMessage->getCommand() )
      {
        case STARTED:
          std::cerr << "started" << std::endl;
          break;
        case TERMINATED:
          std::cerr << "terminated" << std::endl;
          break;
        case SUSPENDED:
          std::cerr << "suspended" << std::endl;
          break;
        case RESUMED:
          std::cerr << "resumed" << std::endl;
          break;
        case EVALUATED:
          std::cerr << "evaluated" << std::endl;
          break;
      }
#endif
      theEventSocket->send( lMessage.get(), length );
#ifndef NDEBUG
      std::clog << "[Server Thread] event sent" << std::endl;
#endif
    } catch( SocketException &e ) {
      std::cerr << e.what() << std::endl;
    }
}


void ZorbaDebugger::startedEvent()
{
    StartedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::suspendedEvent( QueryLoc &aLocation, SuspensionCause aCause )
{
    SuspendedEvent lMessage( aLocation, aCause );
    sendEvent( &lMessage );
}

void ZorbaDebugger::resumedEvent()
{
    ResumedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::terminatedEvent()
{
    TerminatedEvent lMessage;
    sendEvent( &lMessage );
}

void ZorbaDebugger::runQuery()
{
  setStatus( QUERY_RUNNING );
  try
  {
    Zorba_SerializerOptions lSerOptions;
    lSerOptions.omit_xml_declaration = ZORBA_OMIT_XML_DECLARATION_YES;
    theQuery->serialize( std::cout, lSerOptions );
    std::cout.flush();
  } catch ( StaticException& se ) {
    std::cerr << se << std::endl;
  } catch ( DynamicException& de ) {
    std::cerr << de << std::endl;
  } catch ( SystemException& se ) {
    std::cerr << se << std::endl;
  }
  setStatus( QUERY_TERMINATED );
}

bool ZorbaDebugger::hasToSuspend()
{
  if( theStatus == QUERY_SUSPENDED )
  {
    return true;
  }
  std::map<unsigned int, QueryLoc>::iterator it;
  //TODO: can be faster
  for ( it = theBreakpoints.begin(); it != theBreakpoints.end(); it++ )
  {
    std::string lCurrentFilename = theLocation.getFilename();
    std::string lBreakFilename = it->second.getFilename();
    int start = lCurrentFilename.length() - lBreakFilename.length();
    bool fileMatch = true;
    if( start >= 0){
      fileMatch = (lBreakFilename == lCurrentFilename.substr(start));
    }
    if (fileMatch && theLocation.getLineno() == it->second.getLineno() )
    {
      setStatus( QUERY_SUSPENDED, CAUSE_BREAKPOINT );
      return true;
    }
  }
 
  std::map<unsigned int, xqpString>::iterator lIter;
  for ( lIter = theWatchpoints.begin(); lIter != theWatchpoints.end(); lIter++ )
  {
    std::auto_ptr< CompilerCB > ccb(new CompilerCB(*thePlanState->theCompilerCB));
    std::auto_ptr< dynamic_context > dctx(new dynamic_context(thePlanState->dctx()));
      
    //TODO: catch exceptions
    std::auto_ptr<PlanWrapperHolder> eval_plan = compileEvalPlan(theLocation, ccb.get(), dctx.get(), lIter->second, *thePlanState);
    PlanWrapper* lIterator = eval_plan->get();
    try
    {
      store::Item_t lItem;
      std::map<xqpString, xqpString> lValuesAndTypes;

      error::ErrorManager lErrorManger;

      lIterator->next(lItem);
      
      if(lItem->getEBV()->getBooleanValue())
      {
        setStatus(QUERY_SUSPENDED, CAUSE_BREAKPOINT);
	      return true;
      }
    } catch ( error::ZorbaError& e) {
      //do nothing... 
    }
  }
  return false;
}

/**
 * Perform an handshake with the client.
 * Simply returns the received first 11 bytes.
 */
void ZorbaDebugger::handshake( TCPSocket * aSock )
{
  ZorbaArrayAutoPointer<Byte> msg(new Byte[11]);
  try
  {
    aSock->recv( msg.get(), 11 );
    aSock->send( msg.get(), 11 );
  } catch( SocketException &e ) {
    std::cerr << e.what() << std::endl;
  }
}

/**
 * TODO: check if the client has closed the connection
 */
void ZorbaDebugger::handleTcpClient( TCPSocket * aSock )
{
  ZorbaArrayAutoPointer<Byte> lByteMessage;
  std::auto_ptr<AbstractMessage> lMessage;
  std::auto_ptr<ReplyMessage> lReplyMessage;
  Length length;
  try
  {
    lMessage.reset(MessageFactory::buildMessage( aSock ));
    AbstractCommandMessage * lCommandMessage = dynamic_cast< AbstractCommandMessage * > ( lMessage.get() );
    if ( lCommandMessage )
    {
      //process the command message
      //TODO: flush the output somewhere else
      std::cout.flush();
      processMessage( lCommandMessage );
      //Send the reply message
      lReplyMessage.reset(lCommandMessage->getReplyMessage());
      lByteMessage.reset(lReplyMessage->serialize( length ));
      aSock->send( lByteMessage.get(), length );
    } else {
#ifndef NDEBUG
      std::cerr << "Received something wrong" << std::endl;
#endif
      //If something goes wrong, buildMessage() receive a Reply Message containing the error description
      //Send it back to the client right away
      lReplyMessage.reset(dynamic_cast<ReplyMessage *> ( lMessage.get() ));
      if( lReplyMessage.get() ){
        lByteMessage.reset(lReplyMessage->serialize( length ));
        aSock->send( lByteMessage.get(), length );
      } else {
        std::cerr << "Internal error occured. Couldn't send the error message." << std::endl;
      }
    }    
  } catch ( std::exception &e ) {
    std::cerr <<  e.what() << std::endl;
  }
}

void ZorbaDebugger::run()
{
  theRuntimeThread = new Thread( runtimeThread, this );
}

void ZorbaDebugger::suspend()
{
  setStatus( QUERY_SUSPENDED );
}

void ZorbaDebugger::resume()
{
  if ( theStatus == QUERY_SUSPENDED )
  {
    setStatus( QUERY_RUNNING );
    theRuntimeThread->resume();
  }
}

void ZorbaDebugger::terminate()
{
  setStatus( QUERY_TERMINATED );
}

void ZorbaDebugger::quit()
{
  setStatus( QUERY_QUITED );
}

void ZorbaDebugger::eval( xqpString anExpr )
{
  if ( theStatus != QUERY_IDLE && theStatus != QUERY_TERMINATED )
  {
    std::auto_ptr<EvaluatedEvent> lMsg;
    try {
      std::auto_ptr< CompilerCB > ccb(new CompilerCB(*thePlanState->theCompilerCB));
      std::auto_ptr< dynamic_context > dctx(new dynamic_context(thePlanState->dctx()));
      std::auto_ptr<PlanWrapperHolder> eval_plan = compileEvalPlan(theLocation, ccb.get(), dctx.get(), anExpr, *thePlanState);
      PlanWrapper* lIterator = eval_plan->get();
      assert(lIterator != 0);

      store::Item_t lItem;
      std::map<xqpString, xqpString> lValuesAndTypes;

      error::ErrorManager lErrorManger;
      serializer lSerializer(&lErrorManger);
      lSerializer.set_parameter("omit-xml-declaration", "yes");

      while (lIterator->next(lItem)) {
        std::ostringstream os;
        lSerializer.serialize(lItem, os);
        xqpString lValue = os.str();
        xqpString lType( lItem->getType()->getStringValue() );
        lValuesAndTypes.insert(std::pair<xqpString, xqpString>(lValue, lType));
      }
      lMsg.reset(new EvaluatedEvent(anExpr, lValuesAndTypes));
    } catch ( error::ZorbaError& e) {
      std::stringstream lOutputStream;
      xqpString lDescription = e.theDescription.replace("\\\"", "", "");
      lOutputStream << "Error: " << error::ZorbaError::toString(e.theErrorCode) << " " << lDescription;
      lMsg.reset(new EvaluatedEvent( anExpr, lOutputStream.str() ));
    }
    assert(lMsg.get() != 0);
    sendEvent(lMsg.get());
  }
}

std::auto_ptr<PlanWrapperHolder>
ZorbaDebugger::compileEvalPlan(const QueryLoc& loc, CompilerCB* ccb, dynamic_context* dctx, const xqpString& anExpr, PlanState& planState)
{
  std::auto_ptr<PlanWrapperHolder> eval_plan(new PlanWrapperHolder());
  
  checked_vector< std::string > var_keys;

  //set up eval state's ccb
  ccb->m_sctx_list.push_back( ccb->m_sctx = ccb->m_sctx->create_child_context() );
  ccb->m_debugger = 0;

  eval_plan->reset(
    new PlanWrapper (
      EvalIterator::compile ( ccb, anExpr, theVarnames, theVartypes ),
      ccb,
      dctx,
      planState.theStackDepth + 1 )
    );
  eval_plan->get()->checkDepth( loc );
  
  for (unsigned i = 0; i < theChildren.size () - 1; i++)
  {
    //reset the plan iterator for multiple execution
    theChildren[i+1]->reset(*thePlanState);
    store::Iterator_t lIter = new PlanIteratorWrapper(theChildren[i+1], *thePlanState);
    // TODO: is saving an open iterator efficient?
    // Then again if we close theChildren [1] here,
    // we won't be able to re-open it later via the PlanIteratorWrapper
    dctx->add_variable(dynamic_context::var_key(ccb->m_sctx->lookup_var(theVarnames[i])), lIter);
  }
  return std::auto_ptr<PlanWrapperHolder>(eval_plan.release());
}

void ZorbaDebugger::processMessage(AbstractCommandMessage * aMessage)
{
  switch( aMessage->getCommandSet() )
  {
    case EXECUTION:
    {
      switch( aMessage->getCommand() )
      {
        case RUN:
        {
#ifndef NDEBUG
          RunMessage * lMessage = dynamic_cast<RunMessage *> ( aMessage );
          assert( lMessage );
#endif
          run();
          break;
        }
        case SUSPEND:
        {
#ifndef NDEBUG
          SuspendMessage * lMessage = dynamic_cast<SuspendMessage *> ( aMessage );
          assert( lMessage );
#endif
          suspend();
          break;
        }
        case RESUME:
        {
#ifndef NDEBUG
          ResumeMessage * lMessage = dynamic_cast<ResumeMessage*> ( aMessage );
          assert( lMessage );
#endif
          resume();
          break;
        }
        case QUIT:
        {
#ifndef NDEGUB
          QuitMessage * lMessage = dynamic_cast<QuitMessage *>( aMessage );
          assert( lMessage );
#endif
          quit();
          break;
        }
        case TERMINATE:
        {
#ifndef NDEBUG
          TerminateMessage * lMessage = dynamic_cast<TerminateMessage *> ( aMessage );
          assert( lMessage );
#endif
          terminate();
          break;
        }
        case STEP:
        {
#ifndef NDEBUG
          StepMessage * lMessage = dynamic_cast<StepMessage *> ( aMessage );
          assert( lMessage );
          processMessage( lMessage );
#else
          processMessage( static_cast<StepMessage *> ( aMessage ));
#endif
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for execution command set .");
    }
      break;
    case BREAKPOINTS:
    {
      switch( aMessage->getCommand() )
      {
        case SET:
        {
          SetMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< SetMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage =  static_cast< SetMessage * > ( aMessage );
#endif
          std::map<unsigned int, QueryLoc> locations = lMessage->getLocations();
          std::map<unsigned int, QueryLoc>::iterator it;
          for ( it = locations.begin(); it != locations.end(); it++ )
          {
            theBreakpoints.insert( std::make_pair( it->first, it->second ) ); 
          }
          
          std::map<unsigned int, xqpString> exprs = lMessage->getExprs();
          std::map<unsigned int, xqpString>::iterator lIt;
          for ( lIt = exprs.begin(); lIt != exprs.end(); lIt++ )
          {
            theWatchpoints.insert( std::make_pair( lIt->first, lIt->second ) ); 
          }
          break;
        }
        case CLEAR:
        {
          ClearMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< ClearMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage =  static_cast< ClearMessage * > ( aMessage );
#endif
          std::vector<unsigned int>::iterator it;
          std::vector<unsigned int> ids = lMessage->getIds();
          for ( it = ids.begin(); it != ids.end(); it++ )
          {
            std::map<unsigned int, QueryLoc>::iterator
            lIter = theBreakpoints.find( *it );
            if ( lIter !=  theBreakpoints.end() )
            {
              theBreakpoints.erase( lIter );
            } else {
              std::map<unsigned int, xqpString>::iterator lIter2 =
              theWatchpoints.find( *it );
              if ( lIter2 != theWatchpoints.end() ) {
                theWatchpoints.erase( lIter2 );
              }
            }
          }
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for breakpoints command set.");
      }
    }
      break;
    case ENGINE_EVENT:
    case STATIC:
    case DYNAMIC:
    {
      switch ( aMessage->getCommand() )
      {
        case EVAL:
        {
          EvalMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< EvalMessage * > ( aMessage );
          assert( lMessage );
#else
          lMessage = static_cast< EvalMessage * > ( aMessage );
#endif
          eval( lMessage->getExpr() );
          break;
        }
        case VARIABLES:
        {
          VariableMessage * lMessage;
#ifndef NDEBUG
          lMessage = dynamic_cast< VariableMessage * > ( aMessage );
#else
          lMessage = static_cast< VariableMessage * > ( aMessage );
#endif
          VariableReply * lReply = new VariableReply( lMessage->getId(), DEBUGGER_NO_ERROR );
          for( unsigned i = 0; i<theVarnames.size(); i++ )
          {
            xqpString lName(theVarnames.at(i)->getStringValue());
            xqpString lType;
            if( theVartypes.at(i) != 0 )
            {
              lType = theVartypes.at(i)->toString();
            } else {
              lType = "[XS_UNTYPED]";
            }
            checked_vector<global_binding>::iterator it;
            bool is_global = false;
            for ( it = theGlobals.begin(); it != theGlobals.end(); ++it )
            {
              if ( it->first->get_varname() == theVarnames.at(i) )
              {
                lReply->addGlobal( lName, lType );
                is_global = true;
                break;
              }
            }
            if ( ! is_global )
            {
              lReply->addLocal( lName, lType );
            }
          }
          lMessage->setReplyMessage( lReply );
          break;
        }
        default: throw InvalidCommandException("Internal Error. Command not implemented for dynamic command set.");
      }
    }
    break;
    default:
      throw InvalidCommandException("Internal Error. CommandSet not implemented.");
  }
}
};
}//end of namespace


{'fraction_non_alphanumeric': 0.12462413208681865, 'fraction_numerical': 0.0030616150019135095, 'mean_word_length': 2.6685720016044927, 'pattern_counts': {'":': 0, '<': 303, '<?xml version=': 0, '>': 237, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1763298', 'n_tokens_mistral': 11537, 'n_tokens_neox': 11377, 'n_words': 2563}