Add implementations of storage and recordings interfaces

17 files changed, 478 insertions, 2 deletions

diff --git a/p2pvr/.p2config b/p2pvr/.p2config
index 5c820ab..723dd2e 100644
--- a/p2pvr/.p2config
+++ b/p2pvr/.p2config
@@ -5,3 +5,4 @@ common.filelog.path = /tmp/p2daemon.log
 common.filelog.openmode = w
 common.consolelogLevel = 9
 p2pvr.globaldevices.carddaemon = Devices:default -h defiant -p 10001
+p2pvr.storage.root = /tmp/p2pvr/recordings
diff --git a/p2pvr/ice/p2pvr.ice b/p2pvr/ice/p2pvr.ice
index e37f2a0..2c827af 100644
--- a/p2pvr/ice/p2pvr.ice
+++ b/p2pvr/ice/p2pvr.ice
@@ -255,9 +255,15 @@ module P2PVR {
 		idempotent void UpdateEvents(short type);
 	};
 
+	exception StorageException {
+		string path;
+		string message;
+	};
+
 	interface Storage {
-		idempotent string CreateForEventRecording(Schedule sc, DVBSI::Service se, DVBSI::Event ev);
-		idempotent RawDataClient OpenForWrite(string guid);
+		idempotent string CreateForEventRecording(string ext, Schedule sc, DVBSI::Service se, DVBSI::Event ev);
+		idempotent RawDataClient * OpenForWrite(string guid);
+		idempotent void Close(RawDataClient * file);
 		idempotent void Delete(string guid);
 	};
 
diff --git a/p2pvr/lib/dbClient.cpp b/p2pvr/lib/dbClient.cpp
index cdefc9d..c12cb6c 100644
--- a/p2pvr/lib/dbClient.cpp
+++ b/p2pvr/lib/dbClient.cpp
@@ -31,3 +31,19 @@ DatabaseClient::NoRowsFoundException::NoRowsFoundException() :
 {
 }
 
+VariableType
+operator/(const DatabaseClient::SelectPtr & cmd, unsigned int col)
+{
+	HandleAsVariableType vt;
+	(*cmd)[col].apply(vt);
+	return vt.variable;
+}
+
+VariableType
+operator/(const DatabaseClient::SelectPtr & cmd, const std::string & col)
+{
+	HandleAsVariableType vt;
+	(*cmd)[col].apply(vt);
+	return vt.variable;
+}
+
diff --git a/p2pvr/lib/dbClient.h b/p2pvr/lib/dbClient.h
index 6c59301..78a2651 100644
--- a/p2pvr/lib/dbClient.h
+++ b/p2pvr/lib/dbClient.h
@@ -91,5 +91,11 @@ class DatabaseClient : public virtual CommonObjects {
 		void onAllDatasources(const DataSourceCall &) const;
 };
 
+VariableType
+operator/(const DatabaseClient::SelectPtr & cmd, unsigned int col);
+
+VariableType
+operator/(const DatabaseClient::SelectPtr & cmd, const std::string & col);
+
 #endif
 
diff --git a/p2pvr/lib/muxer.cpp b/p2pvr/lib/muxer.cpp
index 886806e..f7014ed 100644
--- a/p2pvr/lib/muxer.cpp
+++ b/p2pvr/lib/muxer.cpp
@@ -1,3 +1,4 @@
+#include <pch.hpp>
 #include "muxer.h"
 #include <misc.h>
 #include <logger.h>
diff --git a/p2pvr/lib/p2Helpers.cpp b/p2pvr/lib/p2Helpers.cpp
index 96fb802..0c1fe37 100644
--- a/p2pvr/lib/p2Helpers.cpp
+++ b/p2pvr/lib/p2Helpers.cpp
@@ -12,6 +12,19 @@ operator<<<Common::DateTime>(VariableType & vt, const Common::DateTime & dt)
 
 template <>
 VariableType &
+operator>><Common::DateTime>(VariableType & vt, Common::DateTime & dt)
+{
+	const boost::posix_time::ptime & date = vt;
+	dt.Year = date.date().year();
+	dt.Month = date.date().month();
+	dt.Day = date.date().day();
+	dt.Hour = date.time_of_day().hours();
+	dt.Minute = date.time_of_day().minutes();
+	return vt;
+}
+
+template <>
+VariableType &
 operator>>(VariableType & vt, short int & v)
 {
 	v = (int)vt;
diff --git a/p2pvr/lib/p2Helpers.h b/p2pvr/lib/p2Helpers.h
index e7732a0..7ea56de 100644
--- a/p2pvr/lib/p2Helpers.h
+++ b/p2pvr/lib/p2Helpers.h
@@ -12,6 +12,10 @@ operator>>(VariableType & vt, T & v)
 	return vt;
 }
 
+template <>
+VariableType &
+operator>><Common::DateTime>(VariableType & vt, Common::DateTime & dt);
+
 template <typename T>
 VariableType &
 operator>>(VariableType & vt, IceUtil::Optional<T> & v)
diff --git a/p2pvr/lib/recordings.cpp b/p2pvr/lib/recordings.cpp
new file mode 100644
index 0000000..1200d76
--- /dev/null
+++ b/p2pvr/lib/recordings.cpp
@@ -0,0 +1,82 @@
+#include <pch.hpp>
+#include "recordings.h"
+#include "resources.h"
+#include <Ice/Ice.h>
+#include <logger.h>
+#include "sqlContainerCreator.h"
+
+ResourceString(Recording_Insert, lib_sql_Recordings_insert_sql);
+ResourceString(Recording_InsertNewId, lib_sql_Recordings_insertNewId_sql);
+ResourceString(Recording_Delete, lib_sql_Recordings_delete_sql);
+ResourceString(Recording_GetStorage, lib_sql_Recordings_getStorage_sql);
+ResourceString(Recording_GetAll, lib_sql_Recordings_getAll_sql);
+
+template<>
+void
+CreateColumns<P2PVR::RecordingPtr>(const ColumnCreator & cc)
+{
+	cc("recordingid", true);
+	cc("storageaddress", false);
+	cc("guid", false);
+	cc("scheduleid", false);
+	cc("title", false);
+	cc("subtitle", false);
+	cc("description", false);
+	cc("starttime", false);
+	cc("duration", false);
+}
+
+template<>
+void
+UnbindColumns(RowState & rs, const P2PVR::RecordingPtr & r)
+{
+	rs.fields[0] >> r->RecordingId;
+	rs.fields[1] >> r->StorageAddress;
+	rs.fields[2] >> r->Guid;
+	rs.fields[3] >> r->ScheduleId;
+	rs.fields[4] >> r->Title;
+	rs.fields[5] >> r->Subtitle;
+	rs.fields[6] >> r->Description;
+	rs.fields[7] >> r->StartTime;
+	rs.fields[8] >> r->Duration;
+}
+
+int
+Recordings::NewRecording(const P2PVR::RecordingPtr & r, const Ice::Current &)
+{
+	Logger()->messagebf(LOG_INFO, "%s: Creating new recording %s at %s", __PRETTY_FUNCTION__, r->Guid, r->StorageAddress);
+	TxHelper tx(this);
+	auto insert = Modify(Recording_Insert,
+			r->StorageAddress, r->Guid, r->ScheduleId, r->Title, r->Subtitle, r->Description, r->StartTime, r->Duration);
+	insert->execute();
+	r->RecordingId = SelectScalar<int>(Recording_InsertNewId);
+	Logger()->messagebf(LOG_INFO, "%s: Created recording Id: %d", __PRETTY_FUNCTION__, r->RecordingId);
+	return r->RecordingId;
+}
+
+void
+Recordings::DeleteRecording(int id, const Ice::Current & ice)
+{
+	Logger()->messagebf(LOG_INFO, "%s: Deleting recording Id: %d", __PRETTY_FUNCTION__, id);
+	auto ic = ice.adapter->getCommunicator();
+	TxHelper tx(this);
+	auto recordingStorages = Select(Recording_GetStorage, id);
+	while (recordingStorages->fetch()) {
+		std::string addr = recordingStorages / "storageaddress";
+		std::string guid = recordingStorages / "guid";
+		auto storage = P2PVR::StoragePrx::checkedCast(ic->stringToProxy(addr));
+		storage->Delete(guid);
+		Logger()->messagebf(LOG_DEBUG, "%s: Delete %s from StorageAddress %s", __PRETTY_FUNCTION__, guid, addr);
+	}
+	Modify(Recording_Delete, id)->execute();
+}
+
+P2PVR::RecordingList
+Recordings::GetRecordings(const Ice::Current &)
+{
+	P2PVR::RecordingList rtn;
+	SqlContainerCreator<P2PVR::RecordingList, P2PVR::Recording> cc(rtn);
+	cc.populate(Select(Recording_GetAll));
+	return rtn;
+}
+
diff --git a/p2pvr/lib/recordings.h b/p2pvr/lib/recordings.h
new file mode 100644
index 0000000..0f18a36
--- /dev/null
+++ b/p2pvr/lib/recordings.h
@@ -0,0 +1,16 @@
+#ifndef RECORDINGS_H
+#define RECORDINGS_H
+
+#include <p2pvr.h>
+#include <string>
+#include "dbClient.h"
+
+class Recordings : public DatabaseClient, public P2PVR::Recordings {
+	public:
+		int NewRecording(const P2PVR::RecordingPtr & rec, const Ice::Current &);
+		void DeleteRecording(int recordingId, const Ice::Current &);
+		P2PVR::RecordingList GetRecordings(const Ice::Current &);
+};
+
+#endif
+
diff --git a/p2pvr/lib/sql/Recordings_delete.sql b/p2pvr/lib/sql/Recordings_delete.sql
new file mode 100644
index 0000000..3395376
--- /dev/null
+++ b/p2pvr/lib/sql/Recordings_delete.sql
@@ -0,0 +1,2 @@
+DELETE FROM recordings
+WHERE recordingId = ?
diff --git a/p2pvr/lib/sql/Recordings_getAll.sql b/p2pvr/lib/sql/Recordings_getAll.sql
new file mode 100644
index 0000000..c0a096c
--- /dev/null
+++ b/p2pvr/lib/sql/Recordings_getAll.sql
@@ -0,0 +1,3 @@
+SELECT recordingId, storageAddress, guid, scheduleId, title, subtitle, description, startTime, duration
+FROM recordings r
+ORDER BY startTime, title, subtitle
diff --git a/p2pvr/lib/sql/Recordings_getStorage.sql b/p2pvr/lib/sql/Recordings_getStorage.sql
new file mode 100644
index 0000000..fdaf58d
--- /dev/null
+++ b/p2pvr/lib/sql/Recordings_getStorage.sql
@@ -0,0 +1,4 @@
+SELECT storageAddress, guid
+FROM recordings
+WHERE recordingId = ?
+
diff --git a/p2pvr/lib/sql/Recordings_insert.sql b/p2pvr/lib/sql/Recordings_insert.sql
new file mode 100644
index 0000000..dfe2ccc
--- /dev/null
+++ b/p2pvr/lib/sql/Recordings_insert.sql
@@ -0,0 +1,2 @@
+INSERT INTO recordings(storageAddress, guid, scheduleId, title, subtitle, description, startTime, duration)
+VALUES(?, ?, ?, ?, ?, ?, ?, ?)
diff --git a/p2pvr/lib/sql/Recordings_insertNewId.sql b/p2pvr/lib/sql/Recordings_insertNewId.sql
new file mode 100644
index 0000000..0583b49
--- /dev/null
+++ b/p2pvr/lib/sql/Recordings_insertNewId.sql
@@ -0,0 +1,2 @@
+SELECT currval('recordings_recordingid_seq');
+
diff --git a/p2pvr/lib/storage.cpp b/p2pvr/lib/storage.cpp
new file mode 100644
index 0000000..9d9345d
--- /dev/null
+++ b/p2pvr/lib/storage.cpp
@@ -0,0 +1,275 @@
+#include <pch.hpp>
+#include "storage.h"
+#include "fileSink.h"
+#include <commonHelpers.h>
+#include <fcntl.h>
+#include <logger.h>
+#include <boost/filesystem/operations.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+#include <boost/lexical_cast.hpp>
+
+namespace fs = boost::filesystem;
+
+fs::path Storage::root;
+fs::path Storage::byAll;
+fs::path Storage::byTitle;
+fs::path Storage::byDate;
+fs::path Storage::byService;
+fs::path Storage::bySchedule;
+
+DECLARE_OPTIONS(Storage, "P2PVR Storage options")
+("p2pvr.storage.root", Options::value(&root, "recordings"),
+ "Root folder in which to store recordings")
+("p2pvr.storage.all", Options::value(&byAll, "all"),
+ "Sub folder in which to store all recordings")
+("p2pvr.storage.bytitle", Options::value(&byTitle, "title"),
+ "Sub folder in which to store by title")
+("p2pvr.storage.bydate", Options::value(&byDate, "date"),
+ "Sub folder in which to store by date")
+("p2pvr.storage.byservice", Options::value(&byService, "service"),
+ "Sub folder in which to store by service")
+("p2pvr.storage.byschedule", Options::value(&bySchedule, "schedule"),
+ "Sub folder in which to store by schedule")
+END_OPTIONS(Storage);
+
+inline static
+fs::path createPath(const fs::path & start)
+{
+	return start;
+}
+
+template <typename Arg>
+inline static
+fs::path
+operator/(const fs::path & lhs, const IceUtil::Optional<Arg> & rhs)
+{
+	if (rhs) {
+		return lhs / *rhs;
+	}
+	return lhs;
+}
+
+template <typename Arg, typename ... Args>
+inline static
+fs::path createPath(const fs::path & start, const Arg & arg, const Args & ... args)
+{
+	auto path = start / arg;
+	return createPath(path, args...);
+}
+
+template <typename Arg, typename ... Args>
+inline static
+std::string firstNotNull(const Arg & arg, const Args & ...)
+{
+	return boost::lexical_cast<std::string>(arg);
+}
+
+template <typename Arg, typename ... Args>
+inline static
+std::string firstNotNull(const IceUtil::Optional<Arg> & arg, const Args & ... args)
+{
+	return arg ? firstNotNull(*arg, args...) : firstNotNull(args...);
+}
+
+template <typename ... Args>
+inline static
+std::string firstNotNull(const std::string & arg, const Args & ...)
+{
+	return arg;
+}
+
+inline static
+std::string
+formatIf(const boost::format & f)
+{
+	return f.str();
+}
+
+template <typename Arg, typename ... Args>
+inline static
+IceUtil::Optional<std::string>
+formatIf(boost::format & f, const IceUtil::Optional<Arg> & arg, const Args & ... args)
+{
+	if (arg) {
+		f % *arg;
+		return formatIf(f, args...);
+	}
+	else {
+		return IceUtil::Optional<std::string>();
+	}
+}
+
+template <typename Arg, typename ... Args>
+inline static
+IceUtil::Optional<std::string>
+formatIf(boost::format & f, const Arg & arg, const Args & ... args)
+{
+	f % arg;
+	return formatIf(f, args...);
+}
+
+template <typename... Args>
+inline static
+IceUtil::Optional<std::string>
+formatIf(const std::string & msgfmt, const Args & ... args)
+{
+	boost::format fmt(msgfmt);
+	return formatIf(fmt, args...);
+}
+
+std::string
+Storage::CreateForEventRecording(const std::string & ext, const P2PVR::SchedulePtr & schedule, const DVBSI::ServicePtr & service, const DVBSI::EventPtr & event, const Ice::Current &)
+{
+	fs::create_directories(root / byAll);
+	auto id = boost::lexical_cast<std::string>(boost::uuids::random_generator()());
+	fs::path path = root / byAll / id;
+	path.replace_extension(ext);
+	auto fd = open(path.string().c_str(), O_WRONLY | O_CREAT | O_EXCL, 0664);
+	if (fd < 0) {
+		Logger()->messagebf(LOG_ERR, "%s: Failed to open file for writing at %s (%d:%s)", __PRETTY_FUNCTION__,
+				path, errno, strerror(errno));
+		throw P2PVR::StorageException(path.string(), strerror(errno));
+	}
+	createSymlinks(ext, id, schedule, service, event);
+	Logger()->messagebf(LOG_INFO, "%s: Created new recording %s", __PRETTY_FUNCTION__, path);
+
+	close(fd);
+	return id;
+}
+
+void
+Storage::createSymlinks(const std::string & ext, const std::string & target, const P2PVR::SchedulePtr & schedule, const DVBSI::ServicePtr & service, const DVBSI::EventPtr & event)
+{
+	// By title, with optional season and episode information
+	createSymlink(ext, target, createPath(root, byTitle,
+				event->Title,
+				formatIf("Season %02d", event->Season),
+				firstNotNull(
+					formatIf("Part %02d of %02d - %s", event->Episode, event->Episodes, event->Subtitle),
+					formatIf("Episode %02d - %s", event->Episode, event->Subtitle),
+					formatIf("Part %02d of %02d - %s", event->Episode, event->Episodes, event->Description),
+					formatIf("Part %02d of %02d", event->Episode, event->Episodes),
+					formatIf("Episode %02d - %s", event->Episode, event->Description),
+					event->Subtitle,
+					event->Description,
+					formatIf("Episode %02d", event->Episode),
+					event->StartTime)));
+	// By date
+	createSymlink(ext, target, createPath(root, byDate,
+				formatIf("%04d-%02d-%02d", event->StartTime.Year, event->StartTime.Month, event->StartTime.Day),
+				firstNotNull(
+					formatIf("%s - %s", event->Title, event->Subtitle),
+					formatIf("%s - %s", event->Title, event->Description),
+					event->Title)));
+	// By service
+	createSymlink(ext, target, createPath(root, byService,
+				service->Name,
+				firstNotNull(
+					formatIf("%s - %s", event->Title, event->Subtitle),
+					formatIf("%s - %s", event->Title, event->Description),
+					event->Title)));
+	// By schedule title
+	createSymlink(ext, target, createPath(root, bySchedule,
+				formatIf("Title: %s", schedule->Title),
+				formatIf("%04d-%02d-%02d", event->StartTime.Year, event->StartTime.Month, event->StartTime.Day),
+				firstNotNull(
+					formatIf("%s - %s", event->Title, event->Subtitle),
+					formatIf("%s - %s", event->Title, event->Description),
+					event->Title)));
+	// By schedule search
+	createSymlink(ext, target, createPath(root, bySchedule,
+				formatIf("Search: %s", schedule->Search),
+				firstNotNull(
+					formatIf("%s - %s", event->Title, event->Subtitle),
+					formatIf("%s - %s", event->Title, event->Description),
+					event->Title)));
+}
+void
+Storage::createSymlink(const std::string & ext, const std::string & target, const fs::path & link)
+{
+	fs::path path = link;
+	path.replace_extension(ext);
+	Logger()->messagebf(LOG_DEBUG, "%s: link(%s) -> target(%s)", __PRETTY_FUNCTION__, path, target);
+	if (fs::exists(path)) {
+		Logger()->messagebf(LOG_WARNING, "%s: symlink already exists %s", __PRETTY_FUNCTION__, path);
+		return;
+	}
+	fs::create_directories(path.parent_path());
+	fs::path relativeTarget;
+	for (fs::path tmp = path.parent_path(); tmp != root; tmp = tmp.parent_path()) {
+		relativeTarget /= "..";
+	}
+	relativeTarget /= byAll;
+	relativeTarget /= target;
+	relativeTarget.replace_extension(ext);
+	fs::create_symlink(relativeTarget, path);
+}
+
+std::string
+Storage::FindExtension(const std::string & id)
+{
+	fs::directory_iterator end;
+	for (fs::directory_iterator itr(root / byAll); itr != end; ++itr) {
+		if (itr->path().stem() == id) {
+			return itr->path().extension().string();
+		}
+	}
+	return "";
+}
+
+P2PVR::RawDataClientPrx
+Storage::OpenForWrite(const std::string & id, const Ice::Current & ice)
+{
+	fs::path path = root / byAll / id;
+	path.replace_extension(FindExtension(id));
+	auto fd = open(path.string().c_str(), O_WRONLY | O_APPEND | O_LARGEFILE);
+	if (fd < 0) {
+		Logger()->messagebf(LOG_ERR, "%s: Failed to open file for reading at %s (%d:%s)", __PRETTY_FUNCTION__,
+				path, errno, strerror(errno));
+		throw P2PVR::StorageException(path.string(), strerror(errno));
+	}
+	auto openFile = OpenFilePtr(new OpenFile(ice.adapter, new FileSink(fd)));
+	openFiles.insert(openFile);
+	return *openFile;
+}
+
+void
+Storage::Close(const P2PVR::RawDataClientPrx & file, const Ice::Current &)
+{
+	openFiles.erase(std::find_if(openFiles.begin(), openFiles.end(), [&file](const OpenFilePtr & of) { return *of == file; }));
+}
+
+void
+Storage::Delete(const std::string & id, const Ice::Current &)
+{
+	fs::path path = root / byAll / id;
+	path.replace_extension(FindExtension(id));
+	Logger()->messagebf(LOG_INFO, "%s: Deleting links to %s", __PRETTY_FUNCTION__, path);
+	DeleteFrom(path, fs::canonical(root));
+}
+
+void
+Storage::DeleteFrom(const fs::path & path, const fs::path & from)
+{
+	Logger()->messagebf(LOG_DEBUG, "%s: Deleting links to %s to %s", __PRETTY_FUNCTION__, path, from);
+	fs::directory_iterator end;
+	for (fs::directory_iterator itr(from); itr != end; ++itr) {
+		if (fs::is_directory(*itr)) {
+			DeleteFrom(path, *itr);
+		}
+		else {
+			boost::system::error_code err;
+			auto link = fs::canonical(*itr, err);
+			if (err || link == path) {
+				Logger()->messagebf(LOG_DEBUG, "%s: deleting %s", __PRETTY_FUNCTION__, *itr);
+				fs::remove(*itr);
+			}
+		}
+	}
+	if (from != root && fs::is_empty(from)) {
+		Logger()->messagebf(LOG_DEBUG, "%s: deleting directory %s", __PRETTY_FUNCTION__, from);
+		fs::remove(from);
+	}
+}
+
diff --git a/p2pvr/lib/storage.h b/p2pvr/lib/storage.h
new file mode 100644
index 0000000..093e00d
--- /dev/null
+++ b/p2pvr/lib/storage.h
@@ -0,0 +1,38 @@
+#ifndef STORAGE_H
+#define STORAGE_H
+
+#include <p2pvr.h>
+#include <options.h>
+#include <string>
+#include <boost/filesystem/path.hpp>
+#include "temporaryIceAdapterObject.h"
+
+class Storage : public P2PVR::Storage {
+	public:
+		std::string CreateForEventRecording(const std::string & ext, const P2PVR::SchedulePtr &, const DVBSI::ServicePtr &, const DVBSI::EventPtr &, const Ice::Current &);
+		P2PVR::RawDataClientPrx OpenForWrite(const std::string &, const Ice::Current &);
+		void Close(const P2PVR::RawDataClientPrx & file, const Ice::Current &);
+		void Delete(const std::string &, const Ice::Current &);
+
+		INITOPTIONS;
+
+	protected:
+		static void createSymlinks(const std::string & ext, const std::string &, const P2PVR::SchedulePtr &, const DVBSI::ServicePtr &, const DVBSI::EventPtr &);
+		static void createSymlink(const std::string & ext, const std::string & target, const boost::filesystem::path & link);
+		static void DeleteFrom(const boost::filesystem::path &, const boost::filesystem::path &);
+		static std::string FindExtension(const std::string & id);
+
+		typedef TemporarayIceAdapterObject<P2PVR::RawDataClient> OpenFile;
+		typedef boost::shared_ptr<OpenFile> OpenFilePtr;
+		typedef std::set<OpenFilePtr> OpenFiles;
+		OpenFiles openFiles;
+
+		static boost::filesystem::path root;
+		static boost::filesystem::path byAll;
+		static boost::filesystem::path byTitle;
+		static boost::filesystem::path byDate;
+		static boost::filesystem::path byService;
+		static boost::filesystem::path bySchedule;
+};
+
+#endif
diff --git a/p2pvr/lib/temporaryIceAdapterObject.h b/p2pvr/lib/temporaryIceAdapterObject.h
index e8b83d8..0b6d65e 100644
--- a/p2pvr/lib/temporaryIceAdapterObject.h
+++ b/p2pvr/lib/temporaryIceAdapterObject.h
@@ -32,6 +32,11 @@ class TemporarayIceAdapterObject {
 			return proxy;
 		}
 
+		bool operator==(const typename Object::ProxyType & pr) const
+		{
+			return pr == proxy;
+		}
+
 	private:
 		Ice::ObjectAdapterPtr adapter;
 		typename Object::ProxyType proxy;