filter/Filter: add method ReadMore()

This allows FilterPCM() to return more data, which some
implementations may need to do (e.g. FFmpeg).

src/filter/Filter.hxx

@@ -42,11 +42,28 @@ public:
 	 *
 	 * @param src the input buffer
 	 * @return the output buffer (will be invalidated by deleting
-	 * this object or any call to Reset(), FilterPCM() or
-	 * Flush()); may be empty if no output is currently available
+	 * this object or any call to Reset(), FilterPCM(), ReadMore()
+	 * or Flush()); may be empty if no output is currently
+	 * available
 	 */
 	virtual std::span<const std::byte> FilterPCM(std::span<const std::byte> src) = 0;
 
+	/**
+	 * Read more result data from the filter.  After each
+	 * FilterPCM() call, this should be called repeatedly until it
+	 * returns an empty span.
+	 *
+	 * Throws on error.
+	 *
+	 * @return the output buffer (will be invalidated by deleting
+	 * this object or any call to Reset(), FilterPCM(), ReadMore()
+	 * or Flush()); may be empty if no output is currently
+	 * available
+	 */
+	virtual std::span<const std::byte> ReadMore() {
+		return {};
+	}
+
 	/**
 	 * Flush pending data and return it.  This should be called
 	 * repeatedly until it returns an empty span.

src/filter/Observer.cxx

@@ -68,6 +68,10 @@ class FilterObserver::Proxy final : public Filter {
 		return filter->FilterPCM(src);
 	}
 
+	std::span<const std::byte> ReadMore() override {
+		return filter->ReadMore();
+	}
+
 	std::span<const std::byte> Flush() override {
 		return filter->Flush();
 	}

src/filter/plugins/FfmpegFilter.cxx

@@ -84,6 +84,12 @@ FfmpegFilter::FilterPCM(std::span<const std::byte> src)
 	return ReadOutput();
 }
 
+std::span<const std::byte>
+FfmpegFilter::ReadMore()
+{
+	return ReadOutput();
+}
+
 std::span<const std::byte>
 FfmpegFilter::Flush()
 {

src/filter/plugins/FfmpegFilter.hxx

@@ -54,6 +54,7 @@ public:
 
 	/* virtual methods from class Filter */
 	std::span<const std::byte> FilterPCM(std::span<const std::byte> src) override;
+	std::span<const std::byte> ReadMore() override;
 	std::span<const std::byte> Flush() override;
 
 private:

src/filter/plugins/TwoFilters.cxx

@@ -21,13 +21,47 @@ TwoFilters::FilterPCM(std::span<const std::byte> src)
 }
 
 std::span<const std::byte>
-TwoFilters::Flush()
+TwoFilters::ReadMore()
 {
-	if (auto result = first->Flush(); !result.empty())
-		/* Flush() output from the first Filter must be
-		   filtered by the second Filter */
+	assert(first);
+	assert(second);
+
+	/* first read all remaining data from the second filter */
+	if (auto result = second->ReadMore(); !result.empty())
+		return result;
+
+	/* now read more data from the first filter and process it
+           with the second filter */
+	if (auto result = first->ReadMore(); !result.empty())
+		/* output from the first Filter must be filtered by
+		   the second Filter */
 		return second->FilterPCM(result);
 
+	/* both filters have been queried and there's no more data */
+	return {};
+}
+
+std::span<const std::byte>
+TwoFilters::Flush()
+{
+	assert(second);
+
+	/* first read all remaining data from the second filter */
+	if (auto result = second->ReadMore(); !result.empty())
+		return result;
+
+	/* now flush the first filter and process it with the second
+           filter */
+	if (first) {
+		if (auto result = first->Flush(); !result.empty())
+			/* output from the first Filter must be
+			   filtered by the second Filter */
+			return second->FilterPCM(result);
+
+		first.reset();
+	}
+
+	/* finally flush the second filter */
 	return second->Flush();
 }
 

src/filter/plugins/TwoFilters.hxx

@@ -29,6 +29,7 @@ public:
 	}
 
 	std::span<const std::byte> FilterPCM(std::span<const std::byte> src) override;
+	std::span<const std::byte> ReadMore() override;
 	std::span<const std::byte> Flush() override;
 };
 

src/output/Source.cxx

@@ -138,6 +138,8 @@ AudioOutputSource::GetChunkData(const MusicChunk &chunk,
 			*replay_gain_serial_p = chunk.replay_gain_serial;
 		}
 
+		/* note: the ReplayGainFilter doesn't have a
+		   ReadMore() method */
 		data = current_replay_gain_filter->FilterPCM(data);
 	}
 
@@ -231,10 +233,18 @@ AudioOutputSource::Fill(Mutex &mutex)
 void
 AudioOutputSource::ConsumeData(size_t nbytes) noexcept
 {
+	assert(filter);
+
 	pending_data = pending_data.subspan(nbytes);
 
-	if (pending_data.empty())
-		DropCurrentChunk();
+	if (pending_data.empty()) {
+		/* give the filter a chance to return more data in
+		   another buffer */
+		pending_data = filter->ReadMore();
+
+		if (pending_data.empty())
+			DropCurrentChunk();
+	}
 }
 
 std::span<const std::byte>

test/run_filter.cxx

@@ -122,8 +122,9 @@ try {
 		if (nbytes == 0)
 			break;
 
-		auto dest = filter->FilterPCM(std::span{buffer}.first(nbytes));
-		output_fd.FullWrite(dest);
+		for (auto dest = filter->FilterPCM(std::span{buffer}.first(nbytes));
+		     !dest.empty(); dest = filter->ReadMore())
+			output_fd.FullWrite(dest);
 	}
 
 	while (true) {