//------------------------------------------------------------------------------
// <copyright file="LqrSMACFileSet.cpp" company="LiquidRobotics">
//
// Copyright (c) 2014 Liquid Robotics, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
//
// This software is associated with a release.  Liquid Robotics makes no
// representation that the design, interfaces or component distribution will
// remain unchanged in future releases.
//
// </copyright>
//
// <summary>
//   Implementation of the FileSet class, which manages files according to a
//   configurable policy.  In general, data can be appended to the file set and
//   the set of files is constrained according to size.
// </summary>
//
// <owner>Mike Cookson</owner>
//------------------------------------------------------------------------------

#include "LqrSMACFileSet.h"
#include "nsextend.h"
#include <stdexcept>
#include <sstream>
#include <boost/algorithm/string.hpp>
#include <boost/filesystem.hpp>
#include <boost/date_time/gregorian/gregorian.hpp>
#include <stdlib.h>
#include <unistd.h>

using namespace std;

namespace LiquidRobotics {
namespace SMACFramework {

IMPLEMENT_INNER_EXCEPTION(FileSet,AlreadyInitialized,runtime_error)
IMPLEMENT_INNER_EXCEPTION(FileSet,NotInitialized,runtime_error)

FileSet::FileSet()
: initialized(false)
, readyToWrite(1)
, queueEmpty(1)
, numBytesEnqueued(0)
, compressionSem(4)
{
}

FileSet::FileSet(const FileSetSettings& _settings)
: initialized(false)
, settings(_settings)
, readyToWrite(1)
, queueEmpty(1)
, numBytesEnqueued(0)
, compressionSem(4)
{
}

FileSet::~FileSet()
{
}

FileSet::FileEntryListPtr FileSet::GetFileList() const
{
    FileEntryListPtr result;
    Lock lock(const_cast<FileSet&>(*this));

    ThrowIfNotInitialized();
    result.reset(new FileEntryList);

    for (FileEntryList::const_iterator iter = feList.begin(); iter != feList.end(); ++iter)
    {
        FileEntryPtr newItem;
        newItem.reset(new FileEntry(**iter));

        result->push_back(newItem);
    }

    return result;
}

const FileSetSettings& FileSet::GetSettings() const
{
    Lock lock(const_cast<FileSet&>(*this));
    return settings;
}

void FileSet::Init()
{
    Lock lock(*this);

    ThrowIfInitialized();
    PopulateFileEntryList();
    MaintainFileEntries();

    try
    {
    	if (settings.GetArchiveCompressionType() != FileSetSettings::ECT_NONE)
    	{
			processCompressionConnection = compressionQueue.SubscribeWorkItemProcessed(boost::bind(&FileSet::ProcessCompression, this, _1, _2));
			compressionQueue.Initialize();
    	}

    	// I could disable the thread if max bytes queued == 0, but I like to use the read timeout for periodically maintaining
    	// the file list
		processWritesConnection = workItemQueue.SubscribeWorkItemProcessed(boost::bind(&FileSet::ProcessWrites, this, _1, _2));
		readTimeoutConnection = workItemQueue.SubscribeReadTimeout(boost::bind(&FileSet::ReadTimeout, this, _1));
		workItemQueue.Initialize();
    }
    catch(...)
    {
    	lock.unlock();

        if (workItemQueue.IsInitialized())
        	workItemQueue.Shutdown();

        if (compressionQueue.IsInitialized())
        	compressionQueue.Shutdown();

        lock.lock();
        ClearState();
        throw;
    }

    initialized = true;
}

void FileSet::ClearState()
{
	// after shutting queues down
	using namespace boost::posix_time;

    feList.clear();
    lastMaintenanceTime = not_a_date_time;
    numBytesEnqueued = 0;
    processWritesConnection.reset();
    readTimeoutConnection.reset();
    readyToWrite.SetCount(1);
    queueEmpty.SetCount(1);
    workFileStm.reset();
    processCompressionConnection.reset();
    compressionSem.SetCount(4);
}

bool FileSet::Initialized() const
{
    Lock lock(const_cast<FileSet&>(*this));
    return initialized;
}

void FileSet::Shutdown()
{
	using namespace boost::posix_time;

	queueEmpty.Wait();

    Lock lock(*this);

    if (initialized)
    {
    	RollFiles(lock);
        initialized = false;
        lock.unlock();

        try         { workItemQueue.Shutdown(); }
        catch(...)  { }

        try			{ compressionQueue.Shutdown(); }
        catch(...)	{ }

        lock.lock();
        ClearState();
    }
    else
    	queueEmpty.Post();
}

void FileSet::Write(const ByteArray& data)
{
    {
        Lock lock(*this);
        ThrowIfNotInitialized();

        if (!data.size())
        	return;

        else if (!settings.GetMaxBytesQueued())
        {
        	ProcessWrites2(data, lock);
        	return;
        }
    }

    QWDataItemPtr dataCopy;
    dataCopy.reset(new QWDataItem);
    dataCopy->bytes.assign(data.begin(), data.end());

    readyToWrite.Wait();

    bool postPending = true;

    try
    {
        Lock lock(*this);

        ThrowIfNotInitialized();
        if (!numBytesEnqueued) queueEmpty.SetCount(0);
        workItemQueue.EnqueueWorkItem(dataCopy);
        numBytesEnqueued += data.size();
        postPending = false;

        if (numBytesEnqueued < settings.GetMaxBytesQueued())
            readyToWrite.Post();
    }
    catch(...)
    {
        if (postPending)
            readyToWrite.Post();

        throw;
    }
}

void FileSet::Write(const string& data)
{
    const ByteArray temp((const ByteArray::byte*) data.data(), data.length());
    Write(temp);
}

void FileSet::WriteLine(const string& data)
{
    stringstream stm;
    stm << data << endl;

    const string temp = stm.str();
    Write(temp);
}

void FileSet::Write(IUserBlobPtr blob)
{
    {
        Lock lock(*this);
        ThrowIfNotInitialized();

        if (!blob || !blob->GetData())
        	throw invalid_argument("User blob is not initialized properly");

        else if (!blob->GetData()->size())
        	return;

        else if (!settings.GetMaxBytesQueued())
        {
        	ProcessWrites2(*blob->GetData(), lock);
        	blob->OnWriteComplete(GetWorkfilePath());
        	return;
        }
    }

    QWDataItemPtr dataCopy;
    dataCopy.reset(new QWDataItem);
    dataCopy->blob = blob;

    readyToWrite.Wait();

    bool postPending = true;

    try
    {
        Lock lock(*this);

        ThrowIfNotInitialized();
        if (!numBytesEnqueued) queueEmpty.SetCount(0);
        workItemQueue.EnqueueWorkItem(dataCopy);
        numBytesEnqueued += blob->GetData()->size();
        postPending = false;

        if (numBytesEnqueued < settings.GetMaxBytesQueued())
            readyToWrite.Post();
    }
    catch(...)
    {
        if (postPending)
            readyToWrite.Post();

        throw;
    }
}


void FileSet::WriteSettings( const FileSetSettings& newValue)
{
    Lock lock(*this);
    ThrowIfInitialized();
}

string FileSet::GetWorkfilePath() const
{
	using namespace boost;
	using namespace boost::filesystem;
	const string prefix = settings.GetPrefix();
	const string ext = settings.GetExtension();

	stringstream fnstm;
	fnstm << prefix;

	if (prefix.length() && ext.length() && !starts_with(ext, "."))
		fnstm << ".";

	fnstm << ext;

	string fileName = fnstm.str();
	string result;

	if (fileName.length())
	{
		path fullPath = absolute(path(settings.GetWorkingDirectory()));

		fullPath /= fileName;
		result = fullPath.string();
	}

	return result;
}

void FileSet::ProcessWrites(QueueWorker<QWDataItemPtr>& sender, QWDataItemPtr item)
{
    Lock lock(*this);

    const ByteArray& itemBA = item->blob ? *item->blob->GetData() : item->bytes;

    if (itemBA.size())
    {
    	FileSetSettings::SizeType temp = numBytesEnqueued;

    	numBytesEnqueued -= itemBA.size();

    	if (!numBytesEnqueued)
    		queueEmpty.Post();

    	if ((numBytesEnqueued < settings.GetMaxBytesQueued()) && (temp >= settings.GetMaxBytesQueued()))
			readyToWrite.Post();
    }

    ProcessWrites2(itemBA, lock);
    if (item->blob)
    	item->blob->OnWriteComplete(GetWorkfilePath());
}

void FileSet::ProcessWrites2(const ByteArray& item, Lock& lock)
{
	if (item.size())
	{
		OFStreamPtr stm = OpenWorkFile();

		if (!stm)
		{
			//File too large. File size is >= File Set Settings max size.
			RollFiles(lock);
			stm = OpenWorkFile();
		}

		if (stm)
		{
			OFStream::streampos curPos = stm->tellp();

			OFStream::streampos newPos = curPos;
			newPos += item.size();

			const bool keepWritesTogether = settings.GetKeepWritesTogether();
			const FileSetSettings::FileSizeType maxSize = settings.GetMaxFileSize();
			bool fileInitiallyTooLarge = (curPos >= maxSize);
			bool writeWouldSplit = !fileInitiallyTooLarge && keepWritesTogether && curPos && (newPos >= maxSize);

			if (fileInitiallyTooLarge || writeWouldSplit)
			{
				RollFiles(lock);
				stm = OpenWorkFile();
				curPos = 0;
				newPos = item.size();
				writeWouldSplit = false;
			}

			if (stm)
			{
				const ByteArray::size_type write1 = writeWouldSplit ? (maxSize-curPos) : item.size();
				const ByteArray::size_type write2 = writeWouldSplit ? (item.size()-write1) : 0;

				stm->write((const char*) &item.front(), write1);

				if (stm->fail())
				{
					stm->close();
					stm.reset();
				}

				if (write2)
				{
					RollFiles(lock);
					stm = OpenWorkFile();

					if (stm)
					{
						stm->write(((const char*) &item.front())+write1, write2);

						if (stm->fail())
						{
							stm->close();
							stm.reset();
						}
					}
				}
			}
		}

		workItemQueue.SetTimeout(stm ? 100 : 60000);		// arbitrarily 100ms between writes flushes the buffer
	}

	PeriodicMaintain();
}

FileSet::OFStreamPtr FileSet::OpenWorkFile()
{
    OFStreamPtr result = workFileStm;

    if (!result)
    {
    	string workFilePath = GetWorkfilePath();

    	if (workFilePath.length())
			try
			{
				workFileStm.reset(new OFStream(workFilePath.c_str(), ios_base::out | ios_base::app | ios_base::ate | ios_base::binary));

				if (workFileStm->fail()) {
					workFileStm.reset();
				}
				else
				{
					streampos curPos = workFileStm->tellp();

					if ((settings.GetPurgeStrategy() == FileSetSettings::EPS_NONE)
					 && (feList.size() >= settings.GetMaxFiles())
					 && (curPos >= settings.GetMaxFileSize()))
					{
						workFileStm.reset();
					}

					result = workFileStm;

					if (!curPos)
					{
						ILockableConcept::Lock lock(firingMutex);
						ByteArray newFileData;
						newFileSignal(*this, workFilePath, newFileData);
						lock.unlock();
						if (newFileData.size())
							workFileStm->write((const char*) &newFileData.front(), newFileData.size());
					}
				}
			}
			catch(...)
			{
				workFileStm.reset();
			}
    }

    return result;
}

void FileSet::ReadTimeout(QueueWorkerBase& sender)
{
    Lock lock(*this);

	if (workFileStm)
	{
		workFileStm->close();
		workFileStm.reset();
	}

	workItemQueue.SetTimeout(60000);
	PeriodicMaintain();
}

void FileSet::RollFiles(Lock& lock)
{
	using namespace boost;
	using namespace boost::filesystem;
	using namespace boost::gregorian;
	using namespace boost::posix_time;

	// if the work file is open, close it and deallocate the file pointer
	if (workFileStm)
	{
		workFileStm->close();
		workFileStm.reset();
	}

	// save a copy of the work file name and path
	const string fullWorkfile = GetWorkfilePath();
	path fullWorkfilePath(fullWorkfile);

	// if the work file exists and it's a regular file ...
	if (exists(fullWorkfile) && is_regular_file(fullWorkfile))
	{
		// save a copy of the archive directory as an absolute path
		string archiveDir = settings.GetArchiveDirectory();

		if (archiveDir.length())
			archiveDir = absolute(path(archiveDir)).string();

		// save a copy of the working directory as an absolute path
		string workFileDir = settings.GetWorkingDirectory();

		if (workFileDir.length())
			workFileDir = absolute(path(workFileDir)).string();

		// use the work directory as the archive directory if the archive
		// directory is not configured
		if (!archiveDir.length())
			archiveDir = workFileDir;

		// save the prefix and extension parameters and ensure that any
		// non-empty extension starts with a dot
		string prefix = settings.GetPrefix();
		string ext = settings.GetExtension();

		if (ext.length() && !starts_with(ext, "."))
			ext.insert(0, ".");

		// format the name of the archive file
		const ptime timeNow = microsec_clock::universal_time();
		const date dateNow = timeNow.date();
		const date::ymd_type ymdNow = dateNow.year_month_day();
		const time_duration todNow = timeNow.time_of_day();
		stringstream fnstm;

		fnstm << prefix
			  << "-"
			  << utos(ymdNow.year,10,4) << utos((unsigned short) ymdNow.month,10,2) << utos(ymdNow.day,10,2)
			  << utos(todNow.hours(),10,2) << utos(todNow.minutes(),10,2) << utos(todNow.seconds(),10,2)
			  << "-"
			  << utos(todNow.fractional_seconds(),10,6)
			  << ext;

		string fn = fnstm.str();

		path fullPath(archiveDir);
		fullPath /= fn;

		//cout << "work=" << fullWorkfilePath.string() << " arch=" << fullPath.string() << endl;
		//cout << "work root=" << fullWorkfilePath.root_directory().string() << " arch root=" << fullPath.root_directory().string() << endl;

		string fullPathStr = fullPath.string();

		// give an opportunity for clients to decide the name of the new file
		{ ILockableConcept::Lock lock(firingMutex); fileVolumeCompleteSignal(*this, fullWorkfile, fullPathStr); }

		// try relinking the original work file to the archive name
		// (doesn't throw exceptions)
		system::error_code ec;
		rename(fullWorkfile, fullPathStr, ec);

		//cout << "value=" << ec.value() << " message=" << ec.message() << endl;

		// if the relinking operation failed because the work file
		// and the archive file are in different filesystems, then
		// copy the work file to the archive file location and then
		// delete the work file
		if (ec && (ec.value() == system::errc::cross_device_link))
		{
			copy_file(fullWorkfile, fullPathStr, ec);
			//cout << "(after copy) value=" << ec.value() << " message=" << ec.message() << endl;

			if (!ec)
			{
				system::error_code ec2;
				remove(fullWorkfile, ec);

				ILockableConcept::Lock lock(firingMutex);
				fileRenameSignal(*this, fullWorkfile, fullPathStr);
			}
		}
		else if (!ec)
		{
			ILockableConcept::Lock lock(firingMutex);
			fileRenameSignal(*this, fullWorkfile, fullPathStr);
		}

		// if there was no resulting error, queue the archive file for
		// compression or finish the operation; if the operations so
		// far have failed, just remove the work file
		if (!ec)
		{
			if (settings.GetArchiveCompressionType() == FileSetSettings::ECT_NONE)
				PostCompression(fullPathStr);
			else
			{
				lock.unlock();
				compressionSem.Wait();
				lock.lock();
				compressionQueue.EnqueueWorkItem(fullPathStr);
			}
		}
		else
		{
			remove(fullWorkfile, ec);
			ILockableConcept::Lock lock(firingMutex);
			fileEraseSignal(*this, fullWorkfile);
		}
	}
}

void FileSet::ThrowIfInitialized() const
{
    if (initialized)
        throw AlreadyInitialized("This FileSet instance has already been initialized");
}

void FileSet::ThrowIfNotInitialized() const
{
    if (!initialized)
        throw NotInitialized("This FileSet instance has not been initialized");
}

void FileSet::MaintainFileEntries()
{
	using namespace boost::filesystem;
	using namespace boost::posix_time;
	using namespace boost::system;

	// expire archives based on retention time
	const ptime timeNow = second_clock::universal_time();
	const FileSetSettings::TimeSpan retainTime = settings.GetMaxRetainTime();

	lastMaintenanceTime = timeNow;

	if (retainTime.total_seconds())
	{
		FileEntryList::iterator nextIt;

		for (FileEntryList::iterator it = feList.begin(); it != feList.end(); it = nextIt)
		{
			nextIt = it;
			++nextIt;

			ptime fileModTime = from_time_t((*it)->modifiedTime);

			//cout << "now=" << to_simple_string(timeNow)
			//	 << " path=" << (*it)->filePath
			//	 << " file=" << (*it)->fileName
			//	 << " mod=" << to_simple_string(fileModTime)
			//	 << endl;

			if (fileModTime < timeNow)
			{
				FileSetSettings::TimeSpan elapsed = timeNow-fileModTime;

				//cout << "elapsed=" << to_simple_string(elapsed) << " retain=" << to_simple_string(retainTime) << endl;

				if (elapsed > retainTime)
				{
					path absPath = (*it)->filePath;
					absPath /= (*it)->fileName;

					error_code ec;

					if (!remove(absPath, ec) || !ec)
						feList.erase(it);

					ILockableConcept::Lock lock(firingMutex);
					fileEraseSignal(*this, absPath.native());
				}
			}
		}
	}

	// expire archives in excess of the number of files allowed
	if (settings.GetPurgeStrategy() == FileSetSettings::EPS_OLDEST)
	{
		const FileSetSettings::SizeType numFiles = settings.GetMaxFiles();
		FileEntryList::iterator nextIt;
		FileEntryList::size_type feListSize = feList.size();

		for (FileEntryList::iterator it = feList.begin(); (feListSize >= numFiles) && (it != feList.end()); it = nextIt)
		{
			nextIt = it;
			++nextIt;

			path absPath = (*it)->filePath;
			absPath /= (*it)->fileName;

			error_code ec;

			if (!remove(absPath, ec) || !ec)
			{
				--feListSize;
				feList.erase(it);
			}

			ILockableConcept::Lock lock(firingMutex);
			fileEraseSignal(*this, absPath.native());
		}
	}
}

FileSet::FileEntryPtr FileSet::GetFileEntry(const string& pathToFile)
{
	using namespace boost::filesystem;
	using namespace boost::system;

	path absPath = absolute(path(pathToFile));
	FileEntryPtr result;

	result.reset(new FileEntry);
	result->filePath = absPath.parent_path().string();
	result->fileName = absPath.filename().string();
	result->fileSize = file_size(absPath);					// throws on error
	result->modifiedTime = last_write_time(absPath);		// throws on error

	return result;
}

typedef pair<time_t,string> FileEntryOrderKeyType;

struct FileEntryOrderKeyTypeComparator
{
	bool operator()(const FileEntryOrderKeyType& k1, const FileEntryOrderKeyType& k2)
	{
		   return (k1.first < k2.first) || (k1.second < k2.second);
	}
};

typedef map<FileEntryOrderKeyType,FileSet::FileEntryPtr,FileEntryOrderKeyTypeComparator> FileEntryByModifiedTime;
typedef boost::shared_ptr<boost::filesystem::directory_iterator> DirIterPtr;

void FileSet::PopulateFileEntryList()
{
#if BOOST_FILESYSTEM_VERSION < 3
	using namespace boost::filesystem3;
#else
	using namespace boost::filesystem;
#endif
	using namespace boost::system;

	feList.clear();

	const string workDir = settings.GetWorkingDirectory();
	error_code ec;
	bool haveWorkDir = is_directory(workDir, ec);
	if (ec) haveWorkDir = false;

	const string archDir = settings.GetArchiveDirectory();
	bool haveArchiveDir = is_directory(archDir, ec);
	if (ec) haveArchiveDir = false;

	if (haveWorkDir || haveArchiveDir)
	{
		FileEntryByModifiedTime fileEntryIndex;
		path archDirPath(haveArchiveDir ? archDir : workDir);
		string prefix = settings.GetPrefix();
		string extension = settings.GetExtension();

		prefix += "-";

		if (extension.length() && !boost::starts_with(extension, "."))
			extension.insert(0, ".");

		string bzip2Extension = extension+".bz2";
		string gzipExtension = extension+".gz";

		DirIterPtr diptr;

		try 		{ diptr.reset(new directory_iterator(archDirPath)); }
		catch(...)	{ }

		directory_iterator dirEnd;
		bool caughtEx = false;

		while(!caughtEx && diptr && (*diptr != dirEnd))
		{
			try
			{
				if ((*diptr)->status().type() == regular_file)
				{
					FileEntryPtr fePtr = GetFileEntry((*diptr)->path().string());
					const bool prefixOK =  !prefix.length() || boost::starts_with(fePtr->fileName, prefix);
					const bool extOK = (!extension.length() || boost::ends_with(fePtr->fileName, extension))
							        || boost::ends_with(fePtr->fileName,bzip2Extension)
							        || boost::ends_with(fePtr->fileName,gzipExtension);

					if (prefixOK && extOK)
					{
						FileEntryOrderKeyType key(fePtr->modifiedTime,fePtr->fileName);
						fileEntryIndex[key] = fePtr;
					}
				}

				++(*diptr);
			}
			catch(exception& ex)
			{
				caughtEx = true;
			}
		}

		for (FileEntryByModifiedTime::iterator it = fileEntryIndex.begin(); it != fileEntryIndex.end(); ++it)
			feList.push_back(it->second);
	}
}

void FileSet::ProcessCompression(StringQType& sender, string item)
{
	Lock lock(*this);
	compressionSem.Post();
	const FileSetSettings::ECompressionType compressionType = settings.GetArchiveCompressionType();

	if (compressionType != FileSetSettings::ECT_NONE)
	{
		string quotedName = QuoteName(item, '\'');
		stringstream cmdLineStm;
		string ext;

		switch(compressionType)
		{
		case FileSetSettings::ECT_BZIP2:

			cmdLineStm << "bzip2 \'" << quotedName << "\'";
			ext = ".bz2";
			break;

		case FileSetSettings::ECT_GZIP:

			cmdLineStm << "gzip \'" << quotedName << "\'";
			ext = ".gz";
			break;

		default:

			break;
		}

		string cmdLine = cmdLineStm.str();

		if (cmdLine.length())
		{
			lock.unlock();
			int systemStatus = ::system(cmdLine.c_str());
			lock.lock();

			if (!WEXITSTATUS(systemStatus))
			{
				string old = item;
				item += ext;

				ILockableConcept::Lock (firingMutex);
				fileRenameSignal(*this, old, item);
			}
		}
	}

	PostCompression(item);
}

void FileSet::PostCompression(const string& item)
{
	bool maintain = false;

	try
	{
		FileEntryPtr newItem = GetFileEntry(item);

		if (newItem)
		{
			feList.push_back(newItem);
			maintain = true;
		}
	}
	catch(...)
	{
	}

	if (maintain)
		MaintainFileEntries();
}

string FileSet::QuoteName(const string& input, char quote)
{
	string result;
	stringstream stm;
	stm << quote;

	const char quote2 = (quote == '\'') ? '\"' : '\'';

	for (string::const_iterator it = input.begin(); it != input.end(); ++it)
	{
		const char c = *it;

		if (c == quote)
			stm << quote << quote2 << quote << quote2 << quote;
		else
			stm << c;
	}

	stm << quote;
	result = stm.str();
	return result;
}

void FileSet::PeriodicMaintain()
{
	using namespace boost::posix_time;

	const ptime timeNow = second_clock::universal_time();
	time_duration elapsed;

	if ((lastMaintenanceTime == not_a_date_time)
	 || ((timeNow > lastMaintenanceTime)
	  && (elapsed = timeNow-lastMaintenanceTime, elapsed.total_seconds() > 60)))

		MaintainFileEntries();
}

FileSet::QueueWorkerWithTimeout::QueueWorkerWithTimeout()
: readTimeout(-1)
{
}

int FileSet::QueueWorkerWithTimeout::GetTimeout() const
{
	return readTimeout;
}

void FileSet::QueueWorkerWithTimeout::SetTimeout(int newValue)
{
	readTimeout = newValue;
}

int FileSet::QueueWorkerWithTimeout::DetermineReadTimeout()
{
    return readTimeout;
}

FileSet::ScopedConnectionPtr FileSet::SubscribeFileEraseSignal(FileEraseSignalType::slot_type slot)
{
	ScopedConnectionPtr result;
	ILockableConcept::Lock lock(firingMutex);
	result.reset(new ScopedConnection(fileEraseSignal.connect(slot)));
	return result;
}

FileSet::ScopedConnectionPtr FileSet::SubscribeFileRenameSignal(FileRenameSignalType::slot_type slot)
{
	ScopedConnectionPtr result;
	ILockableConcept::Lock lock(firingMutex);
	result.reset(new ScopedConnection(fileRenameSignal.connect(slot)));
	return result;
}

FileSet::ScopedConnectionPtr FileSet::SubscribeFileVolumeCompleteSignal(FileVolumeCompleteSignalType::slot_type slot)
{
	ScopedConnectionPtr result;
	ILockableConcept::Lock lock(firingMutex);
	result.reset(new ScopedConnection(fileVolumeCompleteSignal.connect(slot)));
	return result;
}

FileSet::ScopedConnectionPtr FileSet::SubscribeNewFileSignal(NewFileSignalType::slot_type slot)
{
	ScopedConnectionPtr result;
	ILockableConcept::Lock lock(firingMutex);
	result.reset(new ScopedConnection(newFileSignal.connect(slot)));
	return result;
}

ILockableConcept& FileSet::GetFiringMutex()
{
	return firingMutex;
}

}
}
