Logic enhanced, not yet completed (675a9532) · Commits · IA2 / data_importer

src/Client.cpp

+23 −17

Original line number	Diff line number	Diff line
		@@ -215,7 +215,7 @@ void Client::handleUpdateLists()
		{
		DEBUG_STREAM << "Client::handleUpdateLists()" << endl;

		if(readState() != Tango::ALARM && m_protocolManager_sp->hasNewFile())
		if(readState() != Tango::ALARM && m_protocolManager_sp->hasNextFile())
		{
		startResolve();
		}
		@@ -333,37 +333,39 @@ void Client::handleReadResponseBody(const boost::system::error_code& errorCode)
		response_sp->ParseFromArray(&m_readBuff[HEADER_SIZE],
		m_readBuff.size() - HEADER_SIZE);

		if(response_sp->state() == Response::REQUEST_ACCEPTED)
		{
		startReadData(m_protocolManager_sp->processResponse(response_sp));
		FileWrapper::SP fileWrapper_sp =
		m_protocolManager_sp->processResponse(response_sp);

		startReadData(fileWrapper_sp);
		}
		else
		catch(std::logic_error& ec)
		{
		ERROR_STREAM << "Client::handleResponse() "
		<< response_sp->status() << endl;
		WARN_STREAM << "Client::handleReadResponseBody() " << ec.what() << endl;

		writeState(Tango::ALARM);
		writeStatus(response_sp->status());
		}
		//TODO: mark file as failed and try with next next
		}
		catch(std::exception& ec)
		catch(std::runtime_error& ec)
		{
		ERROR_STREAM << "Client::handleResponse() " << ec.what() << endl;
		ERROR_STREAM << "Client::handleReadResponseBody() " << ec.what() << endl;

		writeState(Tango::ALARM);
		writeStatus(ec.what());

		//TODO: stop and set ALARM
		}
		catch(...)
		{
		ERROR_STREAM << "Client::handleResponse() Unknown error" << endl;
		ERROR_STREAM << "Client::handleReadResponseBody() Unknown error" << endl;

		writeState(Tango::ALARM);
		writeStatus("Unknown error");

		//TODO: shit storm happens... stop and set ALARM
		}
		}
		else
		{
		ERROR_STREAM << "Client::handleResponse() " << errorCode.message() << endl;
		ERROR_STREAM << "Client::handleReadResponseBody() " << errorCode.message() << endl;

		writeState(Tango::ALARM);
		writeStatus(errorCode.message());
		@@ -391,12 +393,16 @@ void Client::handleReadData(FileWrapper::SP fileWrapper_sp, std::size_t recvByte
		{
		INFO_STREAM << "Client::handleReadData() transfer complete " << endl;

		if(m_protocolManager_sp->hasNewFile())
		if(m_protocolManager_sp->hasNextFile())
		{
		m_protocolManager_sp->nextFile();

		startWriteRequest();
		}
		else
		{
		startUpdateLists();
		}
		}
		}
		else if(errorCode == boost::asio::error::eof)

src/Client.h

+3 −3

Original line number	Diff line number	Diff line
		@@ -159,10 +159,10 @@ protected:
		//Thread for IO service run scoped pointer
		boost::scoped_ptr<boost::thread> m_thread_sp;

		//First connection timeout
		//Connection write read timeout
		boost::asio::deadline_timer m_resetConnectionTimer;

		//Request response timeout
		//File list update time
		boost::asio::deadline_timer m_listsUpdateTimer;

		//Header size on binary stream
		@@ -189,7 +189,7 @@ protected:
		//Read buffer size
		const boost::uint64_t BUFFER_SIZE = 40960;

		//Buffer for file data read from stream (TODO: unify two buffers)
		//Buffer for file data read from stream
		std::vector<char> m_fileBuff;
		};

src/DBManager.cpp

+21 −1

Original line number	Diff line number	Diff line
		@@ -121,6 +121,27 @@ DBManager::NewFileRowsetSP DBManager::retrieveNewFiles(boost::posix_time::ptime
		return newFileRowset_sp;
		}

		//==============================================================================
		// DBManager::updateNewFilePath()
		//==============================================================================
		void DBManager::updateNewFilePath(std::string storagePath, std::string filePath,
		int fileVersion, std::string fileName) throw(soci::soci_error)
		{
		DEBUG_STREAM << "DBManager::updateNewFilePath()" << endl;

		boost::mutex::scoped_lock lock(m_sessionMutex);

		if(m_mainSession_sp->get_backend() == NULL)
		m_mainSession_sp->reconnect();

		*m_mainSession_sp << "insert into " << m_configuration_sp->getDatabaseSchema()
		<< "." << m_configuration_sp->getDatabaseTable() << " (storage_path, "
		<< "file_path) values (:storagePath, :filePath) where file_version = "
		<< ":fileVersion and file_name like :FileName",
		soci::use(storagePath, "storagePath"), soci::use(filePath, "filePath"),
		soci::use(fileVersion, "fileVersion"), soci::use(fileName, "fileName");
		}

		//==============================================================================
		// DBManager::retrieveLastTimestamp()
		//==============================================================================
		@@ -136,7 +157,6 @@ boost::posix_time::ptime DBManager::retrieveLastTimestamp()

		std::tm tm_time;

		//FIXME: max return null, otherwise coalesce does not work
		*m_auxSession_sp << "select coalesce(max(last_timestamp),'1970-01-01 00:00:00')"
		<< " from "<< m_configuration_sp->getAuxDatabaseSchema()
		<< "." << m_configuration_sp->getAuxDatabaseTimestampTable()

src/DBManager.h

+3 −0

Original line number	Diff line number	Diff line
		@@ -80,6 +80,9 @@ public:
		virtual NewFileRowsetSP retrieveNewFiles(boost::posix_time::ptime)
		throw(soci::soci_error);

		virtual void updateNewFilePath(std::string, std::string, int, std::string)
		throw(soci::soci_error);

		//------------------------------------------------------------------------------
		// [Public] Timestamp methods
		//------------------------------------------------------------------------------

src/PlainClient.cpp

+8 −1

Original line number	Diff line number	Diff line
		@@ -141,7 +141,14 @@ void PlainClient::startWriteRequest()
		boost::bind(&PlainClient::handleWriteRequest, this,
		boost::asio::placeholders::error));
		}
		catch(std::exception& ec)
		catch(std::logic_error& ec)
		{
		WARN_STREAM << "PlainClient::startWriteRequest() " << ec.what() << endl;

		writeState(Tango::ALARM);
		writeStatus(ec.what());
		}
		catch(std::runtime_error& ec)
		{
		ERROR_STREAM << "PlainClient::startWriteRequest() " << ec.what() << endl;