storage/Interface: add virtual method OpenFile()

This should replace most InputStream::Open() calls because it is a
chance to reuse existing resources (e.g. the NFS connection).  No such
optimization is implemented, currently (and the method is not yet used
by anybody).

src/storage/CompositeStorage.cxx

@@ -4,6 +4,7 @@
 #include "CompositeStorage.hxx"
 #include "FileInfo.hxx"
 #include "fs/AllocatedPath.hxx"
+#include "input/InputStream.hxx"
 #include "util/IterableSplitString.hxx"
 #include "util/StringCompare.hxx"
 #include "util/StringSplit.hxx"
@@ -320,3 +321,15 @@ CompositeStorage::MapToRelativeUTF8(std::string_view uri) const noexcept
 
 	return relative_buffer;
 }
+
+InputStreamPtr
+CompositeStorage::OpenFile(std::string_view uri_utf8, Mutex &_mutex)
+{
+	const std::lock_guard<Mutex> lock(mutex);
+
+	auto f = FindStorage(uri_utf8);
+	if (f.directory->storage == nullptr)
+		return nullptr;
+
+	return f.directory->storage->OpenFile(f.uri, _mutex);
+}

src/storage/CompositeStorage.hxx

@@ -127,6 +127,8 @@ public:
 
 	std::string_view MapToRelativeUTF8(std::string_view uri) const noexcept override;
 
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &mutex) override;
+
 private:
 	template<typename T>
 	void VisitMounts(std::string &uri, const Directory &directory,

src/storage/StorageInterface.hxx

@@ -3,6 +3,9 @@
 
 #pragma once
 
+#include "input/Ptr.hxx"
+#include "thread/Mutex.hxx"
+
 #include <memory>
 #include <string>
 #include <string_view>
@@ -69,4 +72,12 @@ public:
 	 */
 	[[nodiscard]] [[gnu::pure]]
 	virtual std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept = 0;
+
+	/**
+	 * Open a file in this storage as an #InputStream.
+	 *
+	 * Throws on error
+	 */
+	[[nodiscard]]
+	virtual InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &mutex) = 0;
 };

src/storage/plugins/CurlStorage.cxx

@@ -6,6 +6,9 @@
 #include "storage/StorageInterface.hxx"
 #include "storage/FileInfo.hxx"
 #include "storage/MemoryDirectoryReader.hxx"
+#include "input/InputStream.hxx"
+#include "input/RewindInputStream.hxx"
+#include "input/plugins/CurlInputPlugin.hxx"
 #include "lib/curl/HttpStatusError.hxx"
 #include "lib/curl/Init.hxx"
 #include "lib/curl/Global.hxx"
@@ -52,6 +55,8 @@ public:
 	[[nodiscard]] std::string MapUTF8(std::string_view uri_utf8) const noexcept override;
 
 	[[nodiscard]] std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept override;
+
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &mutex) override;
 };
 
 std::string
@@ -71,6 +76,12 @@ CurlStorage::MapToRelativeUTF8(std::string_view uri_utf8) const noexcept
 					CurlUnescape(uri_utf8));
 }
 
+InputStreamPtr
+CurlStorage::OpenFile(std::string_view uri_utf8, Mutex &mutex)
+{
+	return input_rewind_open(OpenCurlInputStream(MapUTF8(uri_utf8), {}, mutex));
+}
+
 class BlockingHttpRequest : protected CurlResponseHandler {
 	InjectEvent defer_start;
 

src/storage/plugins/LocalStorage.cxx

@@ -5,6 +5,8 @@
 #include "storage/StoragePlugin.hxx"
 #include "storage/StorageInterface.hxx"
 #include "storage/FileInfo.hxx"
+#include "input/InputStream.hxx"
+#include "input/LocalOpen.hxx"
 #include "fs/FileInfo.hxx"
 #include "fs/AllocatedPath.hxx"
 #include "fs/DirectoryReader.hxx"
@@ -50,6 +52,8 @@ public:
 
 	[[nodiscard]] std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept override;
 
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &mutex) override;
+
 private:
 	[[nodiscard]] AllocatedPath MapFSOrThrow(std::string_view uri_utf8) const;
 };
@@ -112,6 +116,16 @@ LocalStorage::MapToRelativeUTF8(std::string_view uri_utf8) const noexcept
 	return PathTraitsUTF8::Relative(base_utf8, uri_utf8);
 }
 
+InputStreamPtr
+LocalStorage::OpenFile(std::string_view uri_utf8, Mutex &mutex)
+{
+	auto path = MapFS(uri_utf8);
+	if (path == nullptr)
+		return nullptr;
+
+	return OpenLocalInputStream(path, mutex);
+}
+
 StorageFileInfo
 LocalStorage::GetInfo(std::string_view uri_utf8, bool follow)
 {

src/storage/plugins/NfsStorage.cxx

@@ -11,6 +11,7 @@
 #include "lib/nfs/Lease.hxx"
 #include "lib/nfs/Connection.hxx"
 #include "lib/nfs/Glue.hxx"
+#include "input/InputStream.hxx"
 #include "fs/AllocatedPath.hxx"
 #include "thread/Mutex.hxx"
 #include "thread/Cond.hxx"
@@ -94,6 +95,8 @@ public:
 
 	[[nodiscard]] std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept override;
 
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &mutex) override;
+
 	/* virtual methods from NfsLease */
 	void OnNfsConnectionReady() noexcept final {
 		assert(state == State::CONNECTING);
@@ -246,6 +249,14 @@ NfsStorage::MapToRelativeUTF8(std::string_view uri_utf8) const noexcept
 	return PathTraitsUTF8::Relative(base, uri_utf8);
 }
 
+InputStreamPtr
+NfsStorage::OpenFile(std::string_view uri_utf8, Mutex &_mutex)
+{
+	// TODO create NfsInputStream directly
+	auto uri = MapUTF8(uri_utf8);
+	return InputStream::Open(uri.c_str(), _mutex);
+}
+
 static void
 Copy(StorageFileInfo &info, const struct nfs_stat_64 &st) noexcept
 {

src/storage/plugins/SmbclientStorage.cxx

@@ -5,6 +5,7 @@
 #include "storage/StoragePlugin.hxx"
 #include "storage/StorageInterface.hxx"
 #include "storage/FileInfo.hxx"
+#include "input/InputStream.hxx"
 #include "lib/smbclient/Init.hxx"
 #include "lib/smbclient/Context.hxx"
 #include "fs/Traits.hxx"
@@ -63,6 +64,8 @@ public:
 	[[nodiscard]] std::string MapUTF8(std::string_view uri_utf8) const noexcept override;
 
 	[[nodiscard]] std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept override;
+
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &file_mutex) override;
 };
 
 std::string
@@ -113,6 +116,13 @@ SmbclientStorage::GetInfo(std::string_view uri_utf8, [[maybe_unused]] bool follo
 	return ::GetInfo(ctx, mutex, mapped.c_str());
 }
 
+InputStreamPtr
+SmbclientStorage::OpenFile(std::string_view uri_utf8, Mutex &file_mutex)
+{
+	auto uri = MapUTF8(uri_utf8);
+	return InputStream::Open(uri.c_str(), file_mutex);
+}
+
 std::unique_ptr<StorageDirectoryReader>
 SmbclientStorage::OpenDirectory(std::string_view uri_utf8)
 {

src/storage/plugins/UdisksStorage.cxx

@@ -6,6 +6,8 @@
 #include "storage/StoragePlugin.hxx"
 #include "storage/StorageInterface.hxx"
 #include "storage/FileInfo.hxx"
+#include "input/InputStream.hxx"
+#include "input/LocalOpen.hxx"
 #include "lib/fmt/ExceptionFormatter.hxx"
 #include "lib/fmt/RuntimeError.hxx"
 #include "lib/dbus/Glue.hxx"
@@ -113,6 +115,12 @@ public:
 
 	std::string_view MapToRelativeUTF8(std::string_view uri_utf8) const noexcept override;
 
+	InputStreamPtr OpenFile(std::string_view uri_utf8, Mutex &file_mutex) override {
+		MountWait();
+		const auto path = mounted_storage->MapFS(uri_utf8);
+		return OpenLocalInputStream(path, file_mutex);
+	}
+
 private:
 	void SetMountPoint(Path mount_point);
 	void LockSetMountPoint(Path mount_point);

src/storage/plugins/meson.build

@@ -50,6 +50,8 @@ storage_plugins = static_library(
     expat_dep,
     nfs_dep,
     smbclient_dep,
+    input_glue_dep,
+    archive_glue_dep,
   ],
 )
 

test/meson.build

@@ -214,6 +214,7 @@ if enable_database
   executable(
     'run_storage',
     'run_storage.cxx',
+    '../src/TagSave.cxx',
     include_directories: inc,
     dependencies: [
       event_dep,

test/run_storage.cxx

@@ -4,10 +4,21 @@
 #include "cmdline/OptionDef.hxx"
 #include "cmdline/OptionParser.hxx"
 #include "event/Thread.hxx"
+#include "ConfigGlue.hxx"
+#include "tag/Tag.hxx"
 #include "storage/Registry.hxx"
 #include "storage/StorageInterface.hxx"
 #include "storage/FileInfo.hxx"
+#include "input/Init.hxx"
+#include "input/InputStream.hxx"
+#include "input/CondHandler.hxx"
+#include "fs/Path.hxx"
+#include "fs/NarrowPath.hxx"
+#include "event/Thread.hxx"
 #include "net/Init.hxx"
+#include "io/BufferedOutputStream.hxx"
+#include "io/FileDescriptor.hxx"
+#include "io/StdioOutputStream.hxx"
 #include "time/ChronoUtil.hxx"
 #include "time/ISO8601.hxx"
 #include "util/PrintException.hxx"
@@ -15,6 +26,12 @@
 #include "util/StringBuffer.hxx"
 #include "Log.hxx"
 #include "LogBackend.hxx"
+#include "TagSave.hxx"
+#include "config.h"
+
+#ifdef ENABLE_ARCHIVE
+#include "archive/ArchiveList.hxx"
+#endif
 
 #include <memory>
 #include <stdexcept>
@@ -32,9 +49,12 @@ Options:
 Available commands:
   ls URI PATH
   stat URI PATH
+  cat URI PATH
 )";
 
 struct CommandLine {
+	FromNarrowPath config_path;
+
 	bool verbose = false;
 
 	const char *command;
@@ -43,10 +63,12 @@ struct CommandLine {
 };
 
 enum class Option {
+	CONFIG,
 	VERBOSE,
 };
 
 static constexpr OptionDef option_defs[] = {
+	{"config", 0, true, "Load a MPD configuration file"},
 	{"verbose", 'v', false, "Verbose logging"},
 };
 
@@ -58,6 +80,10 @@ ParseCommandLine(int argc, char **argv)
 	OptionParser option_parser(option_defs, argc, argv);
 	while (auto o = option_parser.Next()) {
 		switch (static_cast<Option>(o.index)) {
+		case Option::CONFIG:
+			c.config_path = o.value;
+			break;
+
 		case Option::VERBOSE:
 			c.verbose = true;
 			break;
@@ -74,11 +100,20 @@ ParseCommandLine(int argc, char **argv)
 }
 
 class GlobalInit {
+	const ConfigData config;
 	const ScopeNetInit net_init;
 	EventThread io_thread;
 
+#ifdef ENABLE_ARCHIVE
+	const ScopeArchivePluginsInit archive_plugins_init{config};
+#endif
+
+	const ScopeInputPluginsInit input_plugins_init{config, io_thread.GetEventLoop()};
+
 public:
-	GlobalInit() {
+	GlobalInit(Path config_path)
+		:config(AutoLoadConfigFile(config_path))
+	{
 		io_thread.Start();
 	}
 
@@ -159,13 +194,79 @@ Stat(Storage &storage, const char *path)
 	return EXIT_SUCCESS;
 }
 
+static void
+tag_save(FILE *file, const Tag &tag)
+{
+	StdioOutputStream sos(file);
+	WithBufferedOutputStream(sos, [&](auto &bos){
+		tag_save(bos, tag);
+	});
+}
+
+static void
+WaitReady(InputStream &is, std::unique_lock<Mutex> &lock)
+{
+	CondInputStreamHandler handler;
+	is.SetHandler(&handler);
+
+	handler.cond.wait(lock, [&is]{
+		is.Update();
+		return is.IsReady();
+	});
+
+	is.Check();
+}
+
+static void
+Cat(InputStream &is, std::unique_lock<Mutex> &lock, FileDescriptor out)
+{
+	assert(is.IsReady());
+
+	out.SetBinaryMode();
+
+	if (is.HasMimeType())
+		fprintf(stderr, "MIME type: %s\n", is.GetMimeType());
+
+	/* read data and tags from the stream */
+
+	while (!is.IsEOF()) {
+		if (const auto tag = is.ReadTag()) {
+			fprintf(stderr, "Received a tag:\n");
+			tag_save(stderr, *tag);
+		}
+
+		std::byte buffer[16384];
+		const auto nbytes = is.Read(lock, buffer);
+		if (nbytes == 0)
+			break;
+
+		out.FullWrite({buffer, nbytes});
+	}
+
+	is.Check();
+}
+
+static int
+Cat(Storage &storage, const char *path)
+{
+	Mutex mutex;
+	auto is = storage.OpenFile(path, mutex);
+	assert(is);
+
+	std::unique_lock<Mutex> lock(mutex);
+	WaitReady(*is, lock);
+	Cat(*is, lock, FileDescriptor{STDOUT_FILENO});
+
+	return EXIT_SUCCESS;
+}
+
 int
 main(int argc, char **argv)
 try {
 	const auto c = ParseCommandLine(argc, argv);
 
 	SetLogThreshold(c.verbose ? LogLevel::DEBUG : LogLevel::INFO);
-	GlobalInit init;
+	GlobalInit init{c.config_path};
 
 	if (StringIsEqual(c.command, "ls")) {
 		if (c.args.size() != 2) {
@@ -193,6 +294,19 @@ try {
 					   storage_uri);
 
 		return Stat(*storage, path);
+	} else if (StringIsEqual(c.command, "cat")) {
+		if (c.args.size() != 2) {
+			fputs(usage_text, stderr);
+			return EXIT_FAILURE;
+		}
+
+		const char *const storage_uri = c.args[0];
+		const char *const path = c.args[1];
+
+		auto storage = MakeStorage(init.GetEventLoop(),
+					   storage_uri);
+
+		return Cat(*storage, path);
 	} else {
 		fprintf(stderr, "Unknown command\n\n%s", usage_text);
 		return EXIT_FAILURE;