Merge pull request #993 from TomHarte/PushAudio

Adds a push route for lowpass-filtered audio.

Components/6560/6560.hpp

@@ -435,7 +435,7 @@ template <class BusHandler> class MOS6560 {
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		AudioGenerator audio_generator_;
-		Outputs::Speaker::LowpassSpeaker<AudioGenerator> speaker_;
+		Outputs::Speaker::PullLowpass<AudioGenerator> speaker_;
 
 		Cycles cycles_since_speaker_update_;
 		void update_audio() {

Machines/AmstradCPC/AmstradCPC.cpp

@@ -158,7 +158,7 @@ class AYDeferrer {
 	private:
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		GI::AY38910::AY38910<true> ay_;
-		Outputs::Speaker::LowpassSpeaker<GI::AY38910::AY38910<true>> speaker_;
+		Outputs::Speaker::PullLowpass<GI::AY38910::AY38910<true>> speaker_;
 		HalfCycles cycles_since_update_;
 };
 

Machines/Apple/AppleII/AppleII.cpp

@@ -97,7 +97,7 @@ template <Analyser::Static::AppleII::Target::Model model> class ConcreteMachine:
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		Audio::Toggle audio_toggle_;
-		Outputs::Speaker::LowpassSpeaker<Audio::Toggle> speaker_;
+		Outputs::Speaker::PullLowpass<Audio::Toggle> speaker_;
 		Cycles cycles_since_audio_update_;
 
 		// MARK: - Cards

Machines/Apple/AppleIIgs/AppleIIgs.cpp

@@ -1151,7 +1151,7 @@ class ConcreteMachine:
 		Audio::Toggle audio_toggle_;
 		using AudioSource = Outputs::Speaker::CompoundSource<Apple::IIgs::Sound::GLU, Audio::Toggle>;
 		AudioSource mixer_;
-		Outputs::Speaker::LowpassSpeaker<AudioSource> speaker_;
+		Outputs::Speaker::PullLowpass<AudioSource> speaker_;
 		Cycles cycles_since_audio_update_;
 		Cycles cycles_until_audio_event_;
 		static constexpr int audio_divider = 16;

Machines/Apple/Macintosh/DeferredAudio.hpp

@@ -18,7 +18,7 @@ namespace Macintosh {
 struct DeferredAudio {
 	Concurrency::DeferringAsyncTaskQueue queue;
 	Audio audio;
-	Outputs::Speaker::LowpassSpeaker<Audio> speaker;
+	Outputs::Speaker::PullLowpass<Audio> speaker;
 	HalfCycles time_since_update;
 
 	DeferredAudio() : audio(queue), speaker(audio) {}

Machines/Atari/2600/Bus.hpp

@@ -41,7 +41,7 @@ class Bus {
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		TIASound tia_sound_;
-		Outputs::Speaker::LowpassSpeaker<TIASound> speaker_;
+		Outputs::Speaker::PullLowpass<TIASound> speaker_;
 
 		// joystick state
 		uint8_t tia_input_value_[2] = {0xff, 0xff};

Machines/Atari/ST/AtariST.cpp

@@ -483,7 +483,7 @@ class ConcreteMachine:
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		GI::AY38910::AY38910<false> ay_;
-		Outputs::Speaker::LowpassSpeaker<GI::AY38910::AY38910<false>> speaker_;
+		Outputs::Speaker::PullLowpass<GI::AY38910::AY38910<false>> speaker_;
 		HalfCycles cycles_since_audio_update_;
 
 		JustInTimeActor<DMAController> dma_;

Machines/ColecoVision/ColecoVision.cpp

@@ -381,7 +381,7 @@ class ConcreteMachine:
 		TI::SN76489 sn76489_;
 		GI::AY38910::AY38910<false> ay_;
 		Outputs::Speaker::CompoundSource<TI::SN76489, GI::AY38910::AY38910<false>> mixer_;
-		Outputs::Speaker::LowpassSpeaker<Outputs::Speaker::CompoundSource<TI::SN76489, GI::AY38910::AY38910<false>>> speaker_;
+		Outputs::Speaker::PullLowpass<Outputs::Speaker::CompoundSource<TI::SN76489, GI::AY38910::AY38910<false>>> speaker_;
 
 		std::vector<uint8_t> bios_;
 		std::vector<uint8_t> cartridge_;

Machines/Electron/Electron.cpp

@@ -768,7 +768,7 @@ template <bool has_scsi_bus> class ConcreteMachine:
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		SoundGenerator sound_generator_;
-		Outputs::Speaker::LowpassSpeaker<SoundGenerator> speaker_;
+		Outputs::Speaker::PullLowpass<SoundGenerator> speaker_;
 
 		bool speaker_is_enabled_ = false;
 

Machines/Enterprise/Enterprise.cpp

@@ -703,7 +703,7 @@ template <bool has_disk_controller, bool is_6mhz> class ConcreteMachine:
 
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		Dave::Audio dave_audio_;
-		Outputs::Speaker::LowpassSpeaker<Dave::Audio> speaker_;
+		Outputs::Speaker::PullLowpass<Dave::Audio> speaker_;
 		HalfCycles time_since_audio_update_;
 
 		HalfCycles dave_delay_ = HalfCycles(2);

Machines/MSX/MSX.cpp

@@ -748,7 +748,7 @@ class ConcreteMachine:
 		Audio::Toggle audio_toggle_;
 		Konami::SCC scc_;
 		Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle, Konami::SCC> mixer_;
-		Outputs::Speaker::LowpassSpeaker<Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle, Konami::SCC>> speaker_;
+		Outputs::Speaker::PullLowpass<Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle, Konami::SCC>> speaker_;
 
 		Storage::Tape::BinaryTapePlayer tape_player_;
 		bool tape_player_is_sleeping_ = false;

Machines/MasterSystem/MasterSystem.cpp

@@ -487,7 +487,7 @@ class ConcreteMachine:
 		TI::SN76489 sn76489_;
 		Yamaha::OPL::OPLL opll_;
 		Outputs::Speaker::CompoundSource<decltype(sn76489_), decltype(opll_)> mixer_;
-		Outputs::Speaker::LowpassSpeaker<decltype(mixer_)> speaker_;
+		Outputs::Speaker::PullLowpass<decltype(mixer_)> speaker_;
 		uint8_t opll_detection_word_ = 0xff;
 
 		std::vector<std::unique_ptr<Inputs::Joystick>> joysticks_;

Machines/Oric/Oric.cpp

@@ -84,7 +84,7 @@ namespace Oric {
 using DiskInterface = Analyser::Static::Oric::Target::DiskInterface;
 using Processor = Analyser::Static::Oric::Target::Processor;
 using AY = GI::AY38910::AY38910<false>;
-using Speaker = Outputs::Speaker::LowpassSpeaker<AY>;
+using Speaker = Outputs::Speaker::PullLowpass<AY>;
 
 enum ROM {
 	BASIC10 = 0, BASIC11, Microdisc, Colour

Machines/Sinclair/ZX8081/ZX8081.cpp

@@ -466,7 +466,7 @@ template<bool is_zx81> class ConcreteMachine:
 		Concurrency::DeferringAsyncTaskQueue audio_queue_;
 		using AY = GI::AY38910::AY38910<false>;
 		AY ay_;
-		Outputs::Speaker::LowpassSpeaker<AY> speaker_;
+		Outputs::Speaker::PullLowpass<AY> speaker_;
 		HalfCycles time_since_ay_update_;
 		inline void ay_set_register(uint8_t value) {
 			update_audio();

Machines/Sinclair/ZXSpectrum/ZXSpectrum.cpp

@@ -848,7 +848,7 @@ template<Model model> class ConcreteMachine:
 		GI::AY38910::AY38910<false> ay_;
 		Audio::Toggle audio_toggle_;
 		Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle> mixer_;
-		Outputs::Speaker::LowpassSpeaker<Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle>> speaker_;
+		Outputs::Speaker::PullLowpass<Outputs::Speaker::CompoundSource<GI::AY38910::AY38910<false>, Audio::Toggle>> speaker_;
 
 		HalfCycles time_since_audio_update_;
 		void update_audio() {

Outputs/Speaker/Implementation/LowpassSpeaker.hpp

@@ -1,19 +1,20 @@
 //
-//  FilteringSpeaker.h
+//  LowpassSpeaker.hpp
 //  Clock Signal
 //
 //  Created by Thomas Harte on 15/12/2017.
 //  Copyright 2017 Thomas Harte. All rights reserved.
 //
 
-#ifndef FilteringSpeaker_h
+#ifndef LowpassSpeaker_hpp
-#define FilteringSpeaker_h
+#define LowpassSpeaker_hpp
 
 #include "../Speaker.hpp"
 #include "../../../SignalProcessing/FIRFilter.hpp"
 #include "../../../ClockReceiver/ClockReceiver.hpp"
 #include "../../../Concurrency/AsyncTaskQueue.hpp"
 
+#include <algorithm>
 #include <mutex>
 #include <cstring>
 #include <cmath>
@@ -21,64 +22,8 @@
 namespace Outputs {
 namespace Speaker {
 
-/*!
+template <typename ConcreteT, bool is_stereo> class LowpassBase: public Speaker {
-	The low-pass speaker expects an Outputs::Speaker::SampleSource-derived
-	template class, and uses the instance supplied to its constructor as the
-	source of a high-frequency stream of audio which it filters down to a
-	lower-frequency output.
-*/
-template <typename SampleSource> class LowpassSpeaker: public Speaker {
 	public:
-		LowpassSpeaker(SampleSource &sample_source) : sample_source_(sample_source) {
-			// Propagate an initial volume level.
-			sample_source.set_sample_volume_range(32767);
-		}
-
-		void set_output_volume(float volume) final {
-			// Clamp to the acceptable range, and set.
-			volume = std::min(std::max(0.0f, volume), 1.0f);
-			sample_source_.set_sample_volume_range(int16_t(32767.0f * volume));
-		}
-
-		// Implemented as per Speaker.
-		float get_ideal_clock_rate_in_range(float minimum, float maximum) final {
-			std::lock_guard lock_guard(filter_parameters_mutex_);
-
-			// return twice the cut off, if applicable
-			if(	filter_parameters_.high_frequency_cutoff > 0.0f &&
-				filter_parameters_.input_cycles_per_second >= filter_parameters_.high_frequency_cutoff * 3.0f &&
-				filter_parameters_.input_cycles_per_second <= filter_parameters_.high_frequency_cutoff * 3.0f)
-					return filter_parameters_.high_frequency_cutoff * 3.0f;
-
-			// return exactly the input rate if possible
-			if(	filter_parameters_.input_cycles_per_second >= minimum &&
-				filter_parameters_.input_cycles_per_second <= maximum)
-					return filter_parameters_.input_cycles_per_second;
-
-			// if the input rate is lower, return the minimum
-			if(filter_parameters_.input_cycles_per_second < minimum)
-				return minimum;
-
-			// otherwise, return the maximum
-			return maximum;
-		}
-
-		// Implemented as per Speaker.
-		void set_computed_output_rate(float cycles_per_second, int buffer_size, bool) final {
-			std::lock_guard lock_guard(filter_parameters_mutex_);
-			if(filter_parameters_.output_cycles_per_second == cycles_per_second && size_t(buffer_size) == output_buffer_.size()) {
-				return;
-			}
-
-			filter_parameters_.output_cycles_per_second = cycles_per_second;
-			filter_parameters_.parameters_are_dirty = true;
-			output_buffer_.resize(std::size_t(buffer_size) * (SampleSource::get_is_stereo() ? 2 : 1));
-		}
-
-		bool get_is_stereo() final {
-			return SampleSource::get_is_stereo();
-		}
-
 		/*!
 			Sets the clock rate of the input audio.
 		*/
@@ -107,90 +52,42 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 			filter_parameters_.parameters_are_dirty = true;
 		}
 
-		/*!
-			Schedules an advancement by the number of cycles specified on the provided queue.
-			The speaker will advance by obtaining data from the sample source supplied
-			at construction, filtering it and passing it on to the speaker's delegate if there is one.
-		*/
-		void run_for(Concurrency::DeferringAsyncTaskQueue &queue, const Cycles cycles) {
-			queue.defer([this, cycles] {
-				run_for(cycles);
-			});
-		}
-
 	private:
-		enum class Conversion {
+		float get_ideal_clock_rate_in_range(float minimum, float maximum) final {
-			ResampleSmaller,
+			std::lock_guard lock_guard(filter_parameters_mutex_);
-			Copy,
-			ResampleLarger
-		} conversion_ = Conversion::Copy;
 
-		/*!
+			// Return twice the cut off, if applicable.
-			Advances by the number of cycles specified, obtaining data from the sample source supplied
+			if(	filter_parameters_.high_frequency_cutoff > 0.0f &&
-			at construction, filtering it and passing it on to the speaker's delegate if there is one.
+				filter_parameters_.input_cycles_per_second >= filter_parameters_.high_frequency_cutoff * 3.0f &&
-		*/
+				filter_parameters_.input_cycles_per_second <= filter_parameters_.high_frequency_cutoff * 3.0f)
-		void run_for(const Cycles cycles) {
+					return filter_parameters_.high_frequency_cutoff * 3.0f;
-			const auto delegate = delegate_.load(std::memory_order::memory_order_relaxed);
-			if(!delegate) return;
 
-			const int scale = get_scale();
+			// Return exactly the input rate if possible.
+			if(	filter_parameters_.input_cycles_per_second >= minimum &&
+				filter_parameters_.input_cycles_per_second <= maximum)
+					return filter_parameters_.input_cycles_per_second;
 
-			std::size_t cycles_remaining = size_t(cycles.as_integral());
+			// If the input rate is lower, return the minimum...
-			if(!cycles_remaining) return;
+			if(filter_parameters_.input_cycles_per_second < minimum)
+				return minimum;
 
-			FilterParameters filter_parameters;
+			// ... otherwise, return the maximum.
-			{
+			return maximum;
-				std::lock_guard lock_guard(filter_parameters_mutex_);
-				filter_parameters = filter_parameters_;
-				filter_parameters_.parameters_are_dirty = false;
-				filter_parameters_.input_rate_changed = false;
-			}
-			if(filter_parameters.parameters_are_dirty) update_filter_coefficients(filter_parameters);
-			if(filter_parameters.input_rate_changed) {
-				delegate->speaker_did_change_input_clock(this);
-			}
-
-			switch(conversion_) {
-				case Conversion::Copy:
-					while(cycles_remaining) {
-						const auto cycles_to_read = std::min((output_buffer_.size() - output_buffer_pointer_) / (SampleSource::get_is_stereo() ? 2 : 1), cycles_remaining);
-						sample_source_.get_samples(cycles_to_read, &output_buffer_[output_buffer_pointer_ ]);
-						output_buffer_pointer_ += cycles_to_read * (SampleSource::get_is_stereo() ? 2 : 1);
-
-						// TODO: apply scale.
-
-						// Announce to delegate if full.
-						if(output_buffer_pointer_ == output_buffer_.size()) {
-							output_buffer_pointer_ = 0;
-							did_complete_samples(this, output_buffer_, SampleSource::get_is_stereo());
-						}
-
-						cycles_remaining -= cycles_to_read;
-					}
-				break;
-
-				case Conversion::ResampleSmaller:
-					while(cycles_remaining) {
-						const auto cycles_to_read = std::min((input_buffer_.size() - input_buffer_depth_) / (SampleSource::get_is_stereo() ? 2 : 1), cycles_remaining);
-
-						sample_source_.get_samples(cycles_to_read, &input_buffer_[input_buffer_depth_]);
-						input_buffer_depth_ += cycles_to_read * (SampleSource::get_is_stereo() ? 2 : 1);
-
-						if(input_buffer_depth_ == input_buffer_.size()) {
-							resample_input_buffer(scale);
-						}
-
-						cycles_remaining -= cycles_to_read;
-					}
-				break;
-
-				case Conversion::ResampleLarger:
-					// TODO: input rate is less than output rate.
-				break;
-			}
 		}
 
-		SampleSource &sample_source_;
+		// Implemented as per Speaker.
+		void set_computed_output_rate(float cycles_per_second, int buffer_size, bool) final {
+			std::lock_guard lock_guard(filter_parameters_mutex_);
+			if(filter_parameters_.output_cycles_per_second == cycles_per_second && size_t(buffer_size) == output_buffer_.size()) {
+				return;
+			}
+
+			filter_parameters_.output_cycles_per_second = cycles_per_second;
+			filter_parameters_.parameters_are_dirty = true;
+			output_buffer_.resize(std::size_t(buffer_size) * (is_stereo + 1));
+		}
+
+		// MARK: - Filtering.
 
 		std::size_t output_buffer_pointer_ = 0;
 		std::size_t input_buffer_depth_ = 0;
@@ -233,7 +130,6 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 				high_pass_frequency,
 				SignalProcessing::FIRFilter::DefaultAttenuation);
 
-
 			// Pick the new conversion function.
 			if(	filter_parameters.input_cycles_per_second == filter_parameters.output_cycles_per_second &&
 				filter_parameters.high_frequency_cutoff < 0.0) {
@@ -249,7 +145,7 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 			}
 
 			// Do something sensible with any dangling input, if necessary.
-			const int scale = get_scale();
+			const int scale = static_cast<ConcreteT *>(this)->get_scale();
 			switch(conversion_) {
 				// Neither direct copying nor resampling larger currently use any temporary input.
 				// Although in the latter case that's just because it's unimplemented. But, regardless,
@@ -260,7 +156,7 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 					// Reize the input buffer only if absolutely necessary; if sizing downward
 					// such that a sample would otherwise be lost then output it now. Keep anything
 					// currently in the input buffer that hasn't yet been processed.
-					const size_t required_buffer_size = size_t(number_of_taps) * (SampleSource::get_is_stereo() ? 2 : 1);
+					const size_t required_buffer_size = size_t(number_of_taps) * (is_stereo + 1);
 					if(input_buffer_.size() != required_buffer_size) {
 						if(input_buffer_depth_ >= required_buffer_size) {
 							resample_input_buffer(scale);
@@ -273,7 +169,7 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 		}
 
 		inline void resample_input_buffer(int scale) {
-			if constexpr (SampleSource::get_is_stereo()) {
+			if constexpr (is_stereo) {
 				output_buffer_[output_buffer_pointer_ + 0] = filter_->apply(input_buffer_.data(), 2);
 				output_buffer_[output_buffer_pointer_ + 1] = filter_->apply(input_buffer_.data() + 1, 2);
 				output_buffer_pointer_+= 2;
@@ -284,8 +180,8 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 
 			// Apply scale, if supplied, clamping appropriately.
 			if(scale != 65536) {
-				#define SCALE(x) x = int16_t(std::max(std::min((int(x) * scale) >> 16, 32767), -32768))
+				#define SCALE(x) x = int16_t(std::clamp((int(x) * scale) >> 16, -32768, 32767))
-				if constexpr (SampleSource::get_is_stereo()) {
+				if constexpr (is_stereo) {
 					SCALE(output_buffer_[output_buffer_pointer_ - 2]);
 					SCALE(output_buffer_[output_buffer_pointer_ - 1]);
 				} else {
@@ -297,13 +193,13 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 			// Announce to delegate if full.
 			if(output_buffer_pointer_ == output_buffer_.size()) {
 				output_buffer_pointer_ = 0;
-				did_complete_samples(this, output_buffer_, SampleSource::get_is_stereo());
+				did_complete_samples(this, output_buffer_, is_stereo);
 			}
 
 			// If the next loop around is going to reuse some of the samples just collected, use a memmove to
 			// preserve them in the correct locations (TODO: use a longer buffer to fix that?) and don't skip
 			// anything. Otherwise skip as required to get to the next sample batch and don't expect to reuse.
-			const size_t steps = size_t(step_rate_ + position_error_) * (SampleSource::get_is_stereo() ? 2 : 1);
+			const size_t steps = size_t(step_rate_ + position_error_) * (is_stereo + 1);
 			position_error_ = fmodf(step_rate_ + position_error_, 1.0f);
 			if(steps < input_buffer_.size()) {
 				auto *const input_buffer = input_buffer_.data();
@@ -313,18 +209,190 @@ template <typename SampleSource> class LowpassSpeaker: public Speaker {
 				input_buffer_depth_ -= steps;
 			} else {
 				if(steps > input_buffer_.size()) {
-					sample_source_.skip_samples((steps - input_buffer_.size()) / (SampleSource::get_is_stereo() ? 2 : 1));
+					static_cast<ConcreteT *>(this)->skip_samples((steps - input_buffer_.size()) / (1 + is_stereo));
 				}
 				input_buffer_depth_ = 0;
 			}
 		}
 
+		enum class Conversion {
+			ResampleSmaller,
+			Copy,
+			ResampleLarger
+		} conversion_ = Conversion::Copy;
+
+		bool recalculate_filter_if_dirty() {
+			FilterParameters filter_parameters;
+			{
+				std::lock_guard lock_guard(filter_parameters_mutex_);
+				filter_parameters = filter_parameters_;
+				filter_parameters_.parameters_are_dirty = false;
+				filter_parameters_.input_rate_changed = false;
+			}
+			if(filter_parameters.parameters_are_dirty) update_filter_coefficients(filter_parameters);
+			return filter_parameters.input_rate_changed;
+		}
+
+	protected:
+		void process(size_t length) {
+			const auto delegate = delegate_.load(std::memory_order::memory_order_relaxed);
+			if(!delegate) return;
+
+			const int scale = static_cast<ConcreteT *>(this)->get_scale();
+
+			if(recalculate_filter_if_dirty()) {
+				delegate->speaker_did_change_input_clock(this);
+			}
+
+			switch(conversion_) {
+				case Conversion::Copy:
+					while(length) {
+						const auto samples_to_read = std::min((output_buffer_.size() - output_buffer_pointer_) / (1 + is_stereo), length);
+						static_cast<ConcreteT *>(this)->get_samples(samples_to_read, &output_buffer_[output_buffer_pointer_ ]);
+						output_buffer_pointer_ += samples_to_read * (1 + is_stereo);
+
+						// TODO: apply scale.
+
+						// Announce to delegate if full.
+						if(output_buffer_pointer_ == output_buffer_.size()) {
+							output_buffer_pointer_ = 0;
+							did_complete_samples(this, output_buffer_, is_stereo);
+						}
+
+						length -= samples_to_read;
+					}
+				break;
+
+				case Conversion::ResampleSmaller:
+					while(length) {
+						const auto cycles_to_read = std::min((input_buffer_.size() - input_buffer_depth_) / (1 + is_stereo), length);
+						static_cast<ConcreteT *>(this)->get_samples(cycles_to_read, &input_buffer_[input_buffer_depth_]);
+						input_buffer_depth_ += cycles_to_read * (1 + is_stereo);
+
+						if(input_buffer_depth_ == input_buffer_.size()) {
+							resample_input_buffer(scale);
+						}
+
+						length -= cycles_to_read;
+					}
+				break;
+
+				case Conversion::ResampleLarger:
+					// TODO: input rate is less than output rate.
+				break;
+			}
+		}
+};
+
+/*!
+	Provides a low-pass speaker to which blocks of samples are pushed.
+*/
+template <bool is_stereo> class PushLowpass: public LowpassBase<PushLowpass<is_stereo>, is_stereo> {
+	private:
+		using BaseT = LowpassBase<PushLowpass<is_stereo>, is_stereo>;
+		friend BaseT;
+		using BaseT::process;
+
+		std::atomic<uint16_t> scale_ = 32767;
+		int get_scale() {
+			return scale_;
+		}
+
+		const int16_t *buffer_ = nullptr;
+
+		void skip_samples(size_t count) {
+			buffer_ += count;
+		}
+
+		void get_samples(size_t length, int16_t *target) {
+			memcpy(target, buffer_, length);
+			buffer_ += length;
+		}
+
+	public:
+		void set_output_volume(float volume) final {
+			scale_.store(uint16_t(std::clamp(volume * 65535.0f, 0.0f, 65535.0f)));
+		}
+
+		bool get_is_stereo() final {
+			return is_stereo;
+		}
+
+		/*!
+			Filters and posts onward the provided buffer, on the calling thread.
+		*/
+		void push(const int16_t *buffer, size_t length) {
+			buffer_ = buffer;
+			process(length);
+		}
+};
+
+/*!
+	The low-pass speaker expects an Outputs::Speaker::SampleSource-derived
+	template class, and uses the instance supplied to its constructor as the
+	source of a high-frequency stream of audio which it filters down to a
+	lower-frequency output.
+*/
+template <typename SampleSource> class PullLowpass: public LowpassBase<PullLowpass<SampleSource>, SampleSource::get_is_stereo()> {
+	public:
+		PullLowpass(SampleSource &sample_source) : sample_source_(sample_source) {
+			// Propagate an initial volume level.
+			sample_source.set_sample_volume_range(32767);
+		}
+
+		void set_output_volume(float volume) final {
+			// Clamp to the acceptable range, and set.
+			volume = std::clamp(volume, 0.0f, 1.0f);
+			sample_source_.set_sample_volume_range(int16_t(32767.0f * volume));
+		}
+
+		bool get_is_stereo() final {
+			return SampleSource::get_is_stereo();
+		}
+
+		/*!
+			Schedules an advancement by the number of cycles specified on the provided queue.
+			The speaker will advance by obtaining data from the sample source supplied
+			at construction, filtering it and passing it on to the speaker's delegate if there is one.
+		*/
+		void run_for(Concurrency::DeferringAsyncTaskQueue &queue, const Cycles cycles) {
+			queue.defer([this, cycles] {
+				run_for(cycles);
+			});
+		}
+
+	private:
+		using BaseT = LowpassBase<PullLowpass<SampleSource>, SampleSource::get_is_stereo()>;
+		friend BaseT;
+		using BaseT::process;
+
+		/*!
+			Advances by the number of cycles specified, obtaining data from the sample source supplied
+			at construction, filtering it and passing it on to the speaker's delegate if there is one.
+		*/
+		void run_for(const Cycles cycles) {
+			std::size_t cycles_remaining = size_t(cycles.as_integral());
+			if(!cycles_remaining) return;
+
+			process(cycles_remaining);
+		}
+
+		SampleSource &sample_source_;
+
+		void skip_samples(size_t count) {
+			sample_source_.skip_samples(count);
+		}
+
 		int get_scale() {
 			return int(65536.0 / sample_source_.get_average_output_peak());
-		};
+		}
+
+		void get_samples(size_t length, int16_t *target) {
+			sample_source_.get_samples(length, target);
+		}
 };
 
 }
 }
 
-#endif /* FilteringSpeaker_h */
+#endif /* LowpassSpeaker_hpp */