filter/Filter: add virtual method Flush()

Makefile.am

@@ -1644,7 +1644,7 @@ libfilter_api_a_SOURCES = \
 	src/filter/Observer.cxx src/filter/Observer.hxx \
 	src/filter/FilterPlugin.hxx \
 	src/filter/Prepared.hxx \
-	src/filter/Filter.hxx
+	src/filter/Filter.cxx src/filter/Filter.hxx
 
 libfilter_plugins_a_SOURCES = \
 	src/AudioCompress/config.h \

src/filter/Filter.cxx

+/*
+ * Copyright 2003-2017 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include "Filter.hxx"
+#include "util/ConstBuffer.hxx"
+
+ConstBuffer<void>
+Filter::Flush()
+{
+	return nullptr;
+}

src/filter/Filter.hxx

@@ -62,6 +62,12 @@ public:
 	 * or Reset() call)
 	 */
 	virtual ConstBuffer<void> FilterPCM(ConstBuffer<void> src) = 0;
+
+	/**
+	 * Flush pending data and return it.  This should be called
+	 * repepatedly until it returns nullptr.
+	 */
+	virtual ConstBuffer<void> Flush();
 };
 
 #endif

src/filter/Observer.cxx

@@ -75,6 +75,10 @@ public:
 	ConstBuffer<void> FilterPCM(ConstBuffer<void> src) override {
 		return filter->FilterPCM(src);
 	}
+
+	ConstBuffer<void> Flush() noexcept override {
+		return filter->Flush();
+	}
 };
 
 Filter *

src/filter/plugins/ChainFilterPlugin.cxx

@@ -43,6 +43,11 @@ class ChainFilter final : public Filter {
 
 	std::list<Child> children;
 
+	/**
+	 * The child which will be flushed in the next Flush() call.
+	 */
+	std::list<Child>::iterator flushing = children.end();
+
 public:
 	explicit ChainFilter(AudioFormat _audio_format)
 		:Filter(_audio_format) {}
@@ -54,11 +59,20 @@ public:
 		assert(out_audio_format.IsValid());
 
 		children.emplace_back(name, std::move(filter));
+
+		RewindFlush();
 	}
 
 	/* virtual methods from class Filter */
 	void Reset() noexcept override;
 	ConstBuffer<void> FilterPCM(ConstBuffer<void> src) override;
+	ConstBuffer<void> Flush() override;
+
+private:
+	void RewindFlush() {
+		flushing = children.begin();
+	}
+
 };
 
 class PreparedChainFilter final : public PreparedFilter {
@@ -118,23 +132,46 @@ PreparedChainFilter::Open(AudioFormat &in_audio_format)
 void
 ChainFilter::Reset() noexcept
 {
+	RewindFlush();
+
 	for (auto &child : children)
 		child.filter->Reset();
 }
 
-ConstBuffer<void>
-ChainFilter::FilterPCM(ConstBuffer<void> src)
+template<typename I>
+static ConstBuffer<void>
+ApplyFilterChain(I begin, I end, ConstBuffer<void> src)
 {
-	for (auto &child : children) {
+	for (auto i = begin; i != end; ++i)
 		/* feed the output of the previous filter as input
 		   into the current one */
-		src = child.filter->FilterPCM(src);
-	}
+		src = i->filter->FilterPCM(src);
 
-	/* return the output of the last filter */
 	return src;
 }
 
+ConstBuffer<void>
+ChainFilter::FilterPCM(ConstBuffer<void> src)
+{
+	RewindFlush();
+
+	/* return the output of the last filter */
+	return ApplyFilterChain(children.begin(), children.end(), src);
+}
+
+ConstBuffer<void>
+ChainFilter::Flush()
+{
+	for (auto end = children.end(); flushing != end; ++flushing) {
+		auto data = flushing->filter->Flush();
+		if (!data.IsNull())
+			return ApplyFilterChain(std::next(flushing), end,
+						data);
+	}
+
+	return nullptr;
+}
+
 std::unique_ptr<PreparedFilter>
 filter_chain_new() noexcept
 {

src/output/Source.cxx

@@ -244,3 +244,11 @@ AudioOutputSource::ConsumeData(size_t nbytes) noexcept
 	if (pending_data.empty())
 		pipe.Consume(*std::exchange(current_chunk, nullptr));
 }
+
+ConstBuffer<void>
+AudioOutputSource::Flush()
+{
+	return filter
+		? filter->Flush()
+		: nullptr;
+}

src/output/Source.hxx

@@ -195,6 +195,11 @@ public:
 		pipe.ClearTail(chunk);
 	}
 
+	/**
+	 * Wrapper for Filter::Flush().
+	 */
+	ConstBuffer<void> Flush();
+
 private:
 	void OpenFilter(AudioFormat audio_format,
 			PreparedFilter *prepared_replay_gain_filter,

src/output/Thread.cxx

@@ -362,11 +362,42 @@ AudioOutputControl::InternalPause() noexcept
 	skip_delay = true;
 }
 
+static void
+PlayFull(FilteredAudioOutput &output, ConstBuffer<void> _buffer)
+{
+	auto buffer = ConstBuffer<uint8_t>::FromVoid(_buffer);
+
+	while (!buffer.empty()) {
+		size_t nbytes = output.Play(buffer.data, buffer.size);
+		assert(nbytes > 0);
+
+		buffer.skip_front(nbytes);
+	}
+
+}
+
 inline void
 AudioOutputControl::InternalDrain() noexcept
 {
 	const ScopeUnlock unlock(mutex);
 
+	try {
+		/* flush the filter and play its remaining output */
+
+		while (true) {
+			auto buffer = source.Flush();
+			if (buffer.IsNull())
+				break;
+
+			PlayFull(*output, buffer);
+		}
+	} catch (...) {
+		FormatError(std::current_exception(),
+			    "Failed to flush filter on %s", GetLogName());
+		InternalCloseError(std::current_exception());
+		return;
+	}
+
 	output->Drain();
 }
 

test/run_convert.cxx

@@ -86,6 +86,15 @@ try {
 						   output.size);
 	}
 
+	while (true) {
+		auto output = state.Flush();
+		if (output.IsNull())
+			break;
+
+		gcc_unused ssize_t ignored = write(1, output.data,
+						   output.size);
+	}
+
 	state.Close();
 
 	return EXIT_SUCCESS;