IOManagerDB.cpp

#include "IOManagerDB.h"
#include "MySqlException.h"
#include "ZlibStringCompressor.h"
#include "Sct/Serializable.h"
#include "Sct/SctNames.h"
#include "Sct/XmlStyleStream.h"
#include "Sct/StdExceptionWrapper.h"
#include "Sct/Archive/IONameArchiveFile.h"

#include "SctData/UniqueID.h"
#include "SctData/ScanResult.h"
#include "SctData/TestResult.h"

#include <sqlplus.hh>
#include <custom.hh>
#include <sstream>
#include <boost/date_time/posix_time/posix_time.hpp>

using std::string;
using std::ostringstream;
using boost::posix_time::to_simple_string;
using boost::shared_ptr;
using namespace Sct;

/*sql_create_basic_5(data_struct, 1, 5, 
           string, className, 
           unsigned long, runNumber, 
           unsigned long, scanNumber,  
           string, moduleName, 
           string, startTime);
         string, endtime, 
         string, dataBlob); */

namespace SctArchiving {
  
  IOManagerDB::IOManagerDB() {
    m_compressor = shared_ptr<ZlibStringCompressor>(new ZlibStringCompressor());
    m_connection = shared_ptr<Connection>(new Connection("test_db") );
  }
  
  IOManagerDB::~IOManagerDB() {
  }
  
  IOManagerDB* IOManagerDB::s_man=0;
  
  IOManagerDB& IOManagerDB::instance() {
    if (!s_man) s_man=new IOManagerDB();
    return *s_man;
  }
  
  std::string IOManagerDB::status() const {
    std::ostringstream oss;
    oss << "IOManagerDB";
    if (m_compressor.get()){
      oss << "String compressor buffer size : " 
      << m_compressor->getBufferSize() << std::endl;
    } else {
      oss << "String compressor not initialized" << endl;
    }
    return oss.str();
  }

  Query IOManagerDB::getQuery() const{
    return m_connection->query();
  }

  
  const SctData::ResultHeader& IOManagerDB::getHeader(const Serializable& ob){
    const SctData::ResultHeader* header=0;
    if (ob.getClassName().find("TestResult")!=string::npos) {
      header = & dynamic_cast<const SctData::TestResult&>(ob).getHeader();
    } 
    if (ob.getClassName().find("ScanResult")!=string::npos) {
      header = & dynamic_cast<const SctData::ScanResult&>(ob).getHeader();
    }
    if (header) {
      return *header;
    } else {
      ostringstream oss;
      oss << "Dont know how to find a header for object with class name : " 
      << ob.getClassName() << " which has UniqueID : " << ob.getUniqueID();
      throw IoException(oss.str(), __FILE__, __LINE__);
    }
  }

  void IOManagerDB::prepareInsertion(const Serializable& ob, MysqlQuery& query) const{
    const SctData::ResultHeader& header = getHeader(ob);
    SctData::UniqueID id(header.getUniqueID());

    ostringstream oss;
    XmlStyleOStream out_ad(oss);
    writeImpl(out_ad, ob, false);
    
    query << string("INSERT INTO ") << getTable(ob) << string(" VALUES ('")
      << ob.getClassName() << string("','")
      << id.getRunNumber() << string("','")
      << id.getScanNumber() << string("','")
      << id.getModule() << string("','")
      << to_simple_string(header.getStartTime()) << string("','")
      << to_simple_string(header.getEndTime()) << string("','")
      << m_compressor->compress(oss.str(), getCompressionLevel()) << string("')");
  }

  string IOManagerDB::getTable(const Serializable& ob){
    return "data";
  }

  void IOManagerDB::write(const Serializable& ob, const IOParams* par) const {
    try{
      boost::recursive_mutex::scoped_lock lock (getMutex());
      setReadMode(false);

      MysqlQuery query = getQuery();
      prepareInsertion(ob, query);
      query.execute();

      // convert other exceptions to Sct::Throwable
    }catch(BadQuery& er){
      ostringstream oss;
      oss << "Bad Query error number : " << er.error;
      throw MySqlException(oss.str(), __FILE__, __LINE__);
    }catch (BadConversion& er) {
      ostringstream oss;
      oss  << "Tried to convert \"" << er.data << "\" to a \""
       << er.type_name << "\"." << endl;
      throw MySqlException(oss.str(), __FILE__, __LINE__);
    }catch (Sct::Throwable& e){
      throw;
    }catch(std::exception& e){
      StdExceptionWrapper sew(e);
      throw IoException(sew, __FILE__, __LINE__);
    }
  }
  
  string getQueryMatching(const string& name){
    Archive::IONameArchiveFile ioname(name);
    SctData::UniqueID id(ioname.getUniqueID());
    ostringstream oss;
    oss << " WHERE className = " << ioname.getClassName()
    << " AND runNumber   = " << id.getRunNumber()
    << " AND scanNumber  = " << id.getScanNumber()
    << " AND moduleName  = " << id.getModule();
    return oss.str();
  }
  
  boost::shared_ptr<Serializable> IOManagerDB::read(const string& name, const IOParams* par) const {
    try{
      boost::recursive_mutex::scoped_lock lock (getMutex());
      setReadMode(true);
      getReadVersionMap().clear();
      MysqlQuery query = getQuery();
      query << string("SELECT dataBlob FROM DATA ") << getQueryMatching(name);
      Row row = query.use().fetch_row();
      
      const string compressed (row[0]);
      std::istringstream iss(m_compressor->inflate(compressed));
      XmlStyleIStream in_ad(iss);
      
      return boost::dynamic_pointer_cast<Serializable>(readImpl(in_ad));
      
      // convert other exceptions to Sct::Throwable
    }catch(BadQuery& er){
      ostringstream oss;
      oss << "Bad Query error number : " << er.error;
      throw MySqlException(oss.str(), __FILE__, __LINE__);
    }catch (BadConversion& er) {
      ostringstream oss;
      oss  << "Tried to convert \"" << er.data << "\" to a \""
       << er.type_name << "\"." << endl;
      throw MySqlException(oss.str(), __FILE__, __LINE__);      
    }catch (Sct::Throwable& e) {
      throw;
    }catch(std::exception& e){
      StdExceptionWrapper sew(e);
      throw IoException(sew, __FILE__, __LINE__);
    }
  }
}

---