IOThread.cpp

/*
Copyright 2007, 2008, 2009, 2010, 2011 Instituto de Sistemas e Robotica, Instituto Superior Tecnico

This file is part of MeRMaID.

MeRMaID is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

MeRMaID 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 Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with MeRMaID.  If not, see <http://www.gnu.org/licenses/>.
*/



#include "config.h"

#include "IOThread.hpp"
#include "IOThreadUpdateTask.hpp"
#include "AddStreamReaderTask.hpp"
#include "AddStreamWriterTask.hpp"
#include "WriteTask.hpp"
#include "CloseTask.hpp"
#include <ActiveObjectManagerAce.hpp>
#include <cassert>
#include <iostream>

using namespace mermaid::support::io;
using mermaid::support::activeobject::ActiveObjectManagerAce;

// timeout of 1 second
static const int TIMEOUT_MSEC = 1000;

// number of times per second IOThread::run will be executed
static const int IOTHREAD_UPDATE_RATE = 100;

IOThread::~IOThread()
{
}

IOThread::IOThread() : activeObject_ (ActiveObjectManagerAce::getInstance()->createActiveObject()), streamHandler_ (new StreamHandler)
{
  activeObject_->start();
  
  shared_ptr<Task> ioThreadUpdateTask (new IOThreadUpdateTask (this, IOTHREAD_UPDATE_RATE));
  activeObject_->addTask (ioThreadUpdateTask);
}

void IOThread::addStreamReader (shared_ptr<StreamReader> streamReader)
{
  shared_ptr<Task> addStreamReaderTask (new AddStreamReaderTask (this, streamReader));
  this->addTask (addStreamReaderTask);
}

void IOThread::addStreamWriter (shared_ptr<StreamWriter> streamWriter)
{
  shared_ptr<Task> addStreamWriterTask (new AddStreamWriterTask (this, streamWriter));
  this->addTask (addStreamWriterTask);
}

void IOThread::addWriteOperation (int fileDescriptor, size_t bytesToWrite, const char* data, shared_ptr<StreamOperationHandlerMethodBase> streamOperationHandlerWrite)
{
  shared_ptr<Task> addWriteTask (new WriteTask (this, fileDescriptor, bytesToWrite, data, streamOperationHandlerWrite));
  this->addTask (addWriteTask);
}

void IOThread::addCloseOperation (int fileDescriptor, bool writeEnabled, bool readEnabled)
{
  shared_ptr<Task> addCloseTask (new CloseTask (this, fileDescriptor, writeEnabled, readEnabled));
  this->addTask (addCloseTask);
}

void IOThread::run()
{
  ACE_Time_Value time (0, TIMEOUT_MSEC);
  
  // run the proactor event loop
  proactor_.proactor_run_event_loop (time);
  proactor_.proactor_reset_event_loop();
}

void IOThread::addTask (shared_ptr<Task> task)
{
  activeObject_->addTask (task);
  
  // interrupt the event loop so that this task can be processed with minimum delay
  std::cerr << "Interrupted" << std::endl;
  proactor_.proactor_end_event_loop();
}

ACE_Proactor* IOThread::getProactor()
{
  return &proactor_;
}