StreamHandler.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 "StreamHandler.hpp"
#include <Exception.hpp>
#include <utility>
#include <iostream>
#include <sstream>
#include <cassert>
#include <ace/Message_Block.h>

using namespace mermaid::support::io;

using mermaid::support::errorhandling::Exception;

StreamHandler::~StreamHandler()
{

}

void StreamHandler::associateFileDescriptorWithStreamReader (int fileDescriptor, shared_ptr<StreamReader> streamReader)
{
  streamReaderMap_.insert (std::make_pair (fileDescriptor, streamReader));
}

void StreamHandler::associateFileDescriptorWithStreamWriter (int fileDescriptor, shared_ptr<StreamWriter> streamWriter)
{
  streamWriterMap_.insert (std::make_pair (fileDescriptor, streamWriter));
}

void StreamHandler::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
#ifdef DEBUG
  std::cerr << "handle_write_stream  " << result.message_block().length() << std::endl;
#endif
  int fileDescriptor = result.handle();
  
  if (streamReaderMap_.find (fileDescriptor) == streamReaderMap_.end()) {
    std::ostringstream oss;
    oss << fileDescriptor;
    throw Exception ("StreamHandler::handle_read_stream: File descriptor has no associated StreamReader with file descriptor number: " + oss.str());
  }
  
  // use a copy of data, so we can release message block later
  char* msg = new char[result.message_block().length() ];
  std::copy (result.message_block().base(), result.message_block().base() + result.message_block().length(), msg);
  
  // call StreamReader
  streamReaderMap_[fileDescriptor]->onRead (result.message_block().length(), result.bytes_to_read(), msg, result.success());
  
  result.message_block().release();
}

void StreamHandler::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
{
  int fileDescriptor = result.handle();
  
  if (streamWriterMap_.find (fileDescriptor) == streamWriterMap_.end()) {
    std::ostringstream oss;
    oss << fileDescriptor;
    throw Exception ("StreamHandler::handle_write_stream: File descriptor has no associated StreamWriter with file descriptor number: " + oss.str());
  }
  
  assert (result.act() != 0);
  std::cerr << result.act() << std::endl;
  shared_ptr<StreamOperationHandlerMethodBase>* streamOperationHandlerWritePtr = (shared_ptr<StreamOperationHandlerMethodBase>*) result.act();
  shared_ptr<StreamOperationHandlerMethodBase> streamOperationHandlerWrite = *streamOperationHandlerWritePtr;
  delete streamOperationHandlerWritePtr;
  
  // use a copy of data, so we can release message block later
  char* msg = new char[result.message_block().length() ];
  std::copy (result.message_block().base(), result.message_block().base() + result.message_block().length(), msg);
  
  // call StreamWriter
  streamWriterMap_[fileDescriptor]->onWrite (result.message_block().length(), result.bytes_to_write(), msg, result.success(), streamOperationHandlerWrite);
  
  result.message_block().release();
}

shared_ptr<StreamWriter> StreamHandler::getAssociatedStreamWriter (int fileDescriptor)
{
  assert (streamWriterMap_.find (fileDescriptor) != streamWriterMap_.end());
  return streamWriterMap_[fileDescriptor];
}

void StreamHandler::closeStreamReader (int fileDescriptor)
{
  assert (streamReaderMap_.find (fileDescriptor) != streamReaderMap_.end());
  streamReaderMap_.erase (streamReaderMap_.find (fileDescriptor));
}

void StreamHandler::closeStreamWriter (int fileDescriptor)
{
  assert (streamWriterMap_.find (fileDescriptor) != streamWriterMap_.end());
  streamWriterMap_.erase (streamWriterMap_.find (fileDescriptor));
}