Add first working build.

No useful stuff, just ts and discord clients running on the same tokio engine.

Signed-off-by: Aron Heinecke <aron.heinecke@t-online.de>

.gitignore

@@ -0,0 +1,3 @@
+/target
+.credentials
+/.credentials.toml

Cargo.toml

@@ -0,0 +1,56 @@
+[package]
+name = "voice_bridge"
+version = "0.1.0"
+authors = ["Aron Heinecke <aron.heinecke@t-online.de>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+toml = "0.5"
+serde = { version = "1.0", features = ["derive"] }
+
+# tokio tracing from songbird
+tracing = "0.1"
+tracing-subscriber = "0.2"
+tracing-futures = "0.2"
+
+### TS stuff
+audiopus = "0.2"
+sdl2 = "0.34"
+futures = "0.3"
+# we use the ts slog stuff for the code copied
+slog = "2"
+slog-async = "2"
+slog-perf = "0.2"
+slog-term = "2"
+slog-envlogger = "2"
+# copied error handler
+anyhow = "1"
+
+tokio-stream = "0.1"
+
+[dependencies.tsproto-packets]
+git = "https://github.com/ReSpeak/tsclientlib"
+rev = "3aa03661a2b9c62a6faf05e0f16a94e606c4fdb6"
+
+[dependencies.tsclientlib]
+git = "https://github.com/ReSpeak/tsclientlib"
+#default-features = false
+rev = "3aa03661a2b9c62a6faf05e0f16a94e606c4fdb6"
+
+## discord specific
+
+[dependencies.songbird]
+version = "0.1"
+features = ["driver","gateway"]
+
+[dependencies.serenity]
+version = "0.10"
+features = ["client", "standard_framework", "voice","native_tls_backend"]
+
+## tokio
+
+[dependencies.tokio]
+version = "1.0"
+features = ["macros", "rt-multi-thread","signal"]

README.md

@@ -0,0 +1,12 @@
+# SDL setup
+See [sdl2 crate](https://crates.io/crates/sdl2)
+## windows
+For Rustup users, this folder will be in
+
+    C:\Users\{Your Username}\.rustup\toolchains\{current toolchain}\lib\rustlib\{current toolchain}\lib
+
+Copy SDL2.dll from
+
+    SDL2-devel-2.0.x-VC\SDL2-2.0.x\lib\x64\
+
+into your cargo project, right next to your Cargo.toml

src/discord.rs

@@ -0,0 +1,275 @@
+use std::env;
+use serde::Deserialize;
+use tsclientlib::{ClientId, Connection, DisconnectOptions, Identity, StreamItem};
+use tsproto_packets::packets::{AudioData, CodecType, OutAudio, OutPacket};
+use audiopus::coder::Encoder;
+use futures::prelude::*;
+use sdl2::audio::{AudioCallback, AudioDevice, AudioSpec, AudioSpecDesired, AudioStatus};
+use sdl2::AudioSubsystem;
+
+// This trait adds the `register_songbird` and `register_songbird_with` methods
+// to the client builder below, making it easy to install this voice client.
+// The voice client can be retrieved in any command using `songbird::get(ctx).await`.
+use songbird::SerenityInit;
+
+// Import the `Context` to handle commands.
+use serenity::client::Context;
+
+use serenity::{
+    async_trait,
+    client::{Client, EventHandler},
+    framework::{
+        StandardFramework,
+        standard::{
+            Args, CommandResult,
+            macros::{command, group},
+        },
+    },
+    model::{channel::Message, gateway::Ready},
+    Result as SerenityResult,
+};
+
+pub(crate) struct Handler;
+
+#[async_trait]
+impl EventHandler for Handler {
+    async fn ready(&self, _: Context, ready: Ready) {
+        println!("{} is connected!", ready.user.name);
+    }
+}
+
+#[group]
+#[commands(deafen, join, leave, mute, play, ping, undeafen, unmute)]
+pub struct General;
+
+#[derive(Debug,Deserialize)]
+struct Config {
+    discord_token: String,
+    teamspeak_server: String,
+    teamspeak_identity: String,
+    teamspeak_channel: i32,
+}
+
+#[command]
+#[only_in(guilds)]
+async fn deafen(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    let handler_lock = match manager.get(guild_id) {
+        Some(handler) => handler,
+        None => {
+            check_msg(msg.reply(ctx, "Not in a voice channel").await);
+
+            return Ok(());
+        },
+    };
+
+    let mut handler = handler_lock.lock().await;
+
+    if handler.is_deaf() {
+        check_msg(msg.channel_id.say(&ctx.http, "Already deafened").await);
+    } else {
+        if let Err(e) = handler.deafen(true).await {
+            check_msg(msg.channel_id.say(&ctx.http, format!("Failed: {:?}", e)).await);
+        }
+
+        check_msg(msg.channel_id.say(&ctx.http, "Deafened").await);
+    }
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn join(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let channel_id = guild
+        .voice_states.get(&msg.author.id)
+        .and_then(|voice_state| voice_state.channel_id);
+
+    let connect_to = match channel_id {
+        Some(channel) => channel,
+        None => {
+            check_msg(msg.reply(ctx, "Not in a voice channel").await);
+
+            return Ok(());
+        }
+    };
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    let _handler = manager.join(guild_id, connect_to).await;
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn leave(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+    let has_handler = manager.get(guild_id).is_some();
+
+    if has_handler {
+        if let Err(e) = manager.remove(guild_id).await {
+            check_msg(msg.channel_id.say(&ctx.http, format!("Failed: {:?}", e)).await);
+        }
+
+        check_msg(msg.channel_id.say(&ctx.http, "Left voice channel").await);
+    } else {
+        check_msg(msg.reply(ctx, "Not in a voice channel").await);
+    }
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn mute(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    let handler_lock = match manager.get(guild_id) {
+        Some(handler) => handler,
+        None => {
+            check_msg(msg.reply(ctx, "Not in a voice channel").await);
+
+            return Ok(());
+        },
+    };
+
+    let mut handler = handler_lock.lock().await;
+
+    if handler.is_mute() {
+        check_msg(msg.channel_id.say(&ctx.http, "Already muted").await);
+    } else {
+        if let Err(e) = handler.mute(true).await {
+            check_msg(msg.channel_id.say(&ctx.http, format!("Failed: {:?}", e)).await);
+        }
+
+        check_msg(msg.channel_id.say(&ctx.http, "Now muted").await);
+    }
+
+    Ok(())
+}
+
+#[command]
+async fn ping(context: &Context, msg: &Message) -> CommandResult {
+    check_msg(msg.channel_id.say(&context.http, "Pong!").await);
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn play(ctx: &Context, msg: &Message, mut args: Args) -> CommandResult {
+    let url = match args.single::<String>() {
+        Ok(url) => url,
+        Err(_) => {
+            check_msg(msg.channel_id.say(&ctx.http, "Must provide a URL to a video or audio").await);
+
+            return Ok(());
+        },
+    };
+
+    if !url.starts_with("http") {
+        check_msg(msg.channel_id.say(&ctx.http, "Must provide a valid URL").await);
+
+        return Ok(());
+    }
+
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    if let Some(handler_lock) = manager.get(guild_id) {
+        let mut handler = handler_lock.lock().await;
+
+        let source = match songbird::ytdl(&url).await {
+            Ok(source) => source,
+            Err(why) => {
+                println!("Err starting source: {:?}", why);
+
+                check_msg(msg.channel_id.say(&ctx.http, "Error sourcing ffmpeg").await);
+
+                return Ok(());
+            },
+        };
+
+        handler.play_source(source);
+
+        check_msg(msg.channel_id.say(&ctx.http, "Playing song").await);
+    } else {
+        check_msg(msg.channel_id.say(&ctx.http, "Not in a voice channel to play in").await);
+    }
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn undeafen(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    if let Some(handler_lock) = manager.get(guild_id) {
+        let mut handler = handler_lock.lock().await;
+        if let Err(e) = handler.deafen(false).await {
+            check_msg(msg.channel_id.say(&ctx.http, format!("Failed: {:?}", e)).await);
+        }
+
+        check_msg(msg.channel_id.say(&ctx.http, "Undeafened").await);
+    } else {
+        check_msg(msg.channel_id.say(&ctx.http, "Not in a voice channel to undeafen in").await);
+    }
+
+    Ok(())
+}
+
+#[command]
+#[only_in(guilds)]
+async fn unmute(ctx: &Context, msg: &Message) -> CommandResult {
+    let guild = msg.guild(&ctx.cache).await.unwrap();
+    let guild_id = guild.id;
+    
+    let manager = songbird::get(ctx).await
+        .expect("Songbird Voice client placed in at initialisation.").clone();
+
+    if let Some(handler_lock) = manager.get(guild_id) {
+        let mut handler = handler_lock.lock().await;
+        if let Err(e) = handler.mute(false).await {
+            check_msg(msg.channel_id.say(&ctx.http, format!("Failed: {:?}", e)).await);
+        }
+
+        check_msg(msg.channel_id.say(&ctx.http, "Unmuted").await);
+    } else {
+        check_msg(msg.channel_id.say(&ctx.http, "Not in a voice channel to unmute in").await);
+    }
+
+    Ok(())
+}
+
+/// Checks that a message successfully sent; if not, then logs why to stdout.
+fn check_msg(result: SerenityResult<Message>) {
+    if let Err(why) = result {
+        println!("Error sending message: {:?}", why);
+    }
+}

src/main.rs

@@ -0,0 +1,157 @@
+use std::env;
+use serde::Deserialize;
+use tsclientlib::{ClientId, Connection, DisconnectOptions, Identity, StreamItem};
+use tsproto_packets::packets::{AudioData, CodecType, OutAudio, OutPacket};
+use audiopus::coder::Encoder;
+use futures::prelude::*;
+use sdl2::audio::{AudioCallback, AudioDevice, AudioSpec, AudioSpecDesired, AudioStatus};
+use sdl2::AudioSubsystem;
+use slog::{debug, info, o, Drain, Logger};
+use tokio::sync::mpsc;
+use tokio::task::LocalSet;
+use anyhow::*;
+mod ts_voice;
+mod discord;
+use tsproto_packets::packets::{Direction, InAudioBuf};
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+struct ConnectionId(u64);
+
+// This trait adds the `register_songbird` and `register_songbird_with` methods
+// to the client builder below, making it easy to install this voice client.
+// The voice client can be retrieved in any command using `songbird::get(ctx).await`.
+use songbird::SerenityInit;
+
+// Import the `Context` to handle commands.
+use serenity::client::Context;
+
+use serenity::{
+    async_trait,
+    client::{Client, EventHandler},
+    framework::{
+        StandardFramework,
+        standard::{
+            Args, CommandResult,
+            macros::{command, group},
+        },
+    },
+    model::{channel::Message, gateway::Ready},
+    Result as SerenityResult,
+};
+
+
+#[derive(Debug,Deserialize)]
+struct Config {
+    discord_token: String,
+    teamspeak_server: String,
+    teamspeak_identity: String,
+    teamspeak_channel: i32,
+    /// default 0
+    verbose: i32,
+    /// default 1.0
+    volume: f32,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt::init();
+
+    let config: Config = toml::from_str(&std::fs::read_to_string(".credentials.toml").unwrap()).unwrap();
+    let logger = {
+		let decorator = slog_term::TermDecorator::new().build();
+		let drain = slog_term::CompactFormat::new(decorator).build().fuse();
+		let drain = slog_envlogger::new(drain).fuse();
+		let drain = slog_async::Async::new(drain).build().fuse();
+
+		Logger::root(drain, o!())
+	};
+    
+    let framework = StandardFramework::new()
+        .configure(|c| c
+                   .prefix("~"))
+        .group(&discord::GENERAL_GROUP);
+
+    let mut client = Client::builder(&config.discord_token)
+        .event_handler(discord::Handler)
+        .framework(framework)
+        .register_songbird()
+        .await
+        .expect("Err creating client");
+
+    tokio::spawn(async move {
+        let _ = client.start().await.map_err(|why| println!("Client ended: {:?}", why));
+    });
+
+    let con_id = ConnectionId(0);
+	let local_set = LocalSet::new();
+	let audiodata = ts_voice::start(logger.clone(), &local_set)?;
+
+	let con_config = Connection::build(config.teamspeak_server)
+		.log_commands(config.verbose >= 1)
+		.log_packets(config.verbose >= 2)
+		.log_udp_packets(config.verbose >= 3);
+
+	// Optionally set the key of this client, otherwise a new key is generated.
+	let id = Identity::new_from_str(&config.teamspeak_identity).expect("Can't load identity!");
+	let con_config = con_config.identity(id);
+
+	// Connect
+	let mut con = con_config.connect()?;
+
+	let r = con
+		.events()
+		.try_filter(|e| future::ready(matches!(e, StreamItem::BookEvents(_))))
+		.next()
+		.await;
+	if let Some(r) = r {
+		r?;
+	}
+
+	let (send, mut recv) = mpsc::channel(5);
+	{
+		let mut a2t = audiodata.a2ts.lock().unwrap();
+		a2t.set_listener(send);
+		a2t.set_volume(config.volume);
+		a2t.set_playing(true);
+	}
+
+	loop {
+		let t2a = audiodata.ts2a.clone();
+		let events = con.events().try_for_each(|e| async {
+			if let StreamItem::Audio(packet) = e {
+				let from = ClientId(match packet.data().data() {
+					AudioData::S2C { from, .. } => *from,
+					AudioData::S2CWhisper { from, .. } => *from,
+					_ => panic!("Can only handle S2C packets but got a C2S packet"),
+				});
+				let mut t2a = t2a.lock().unwrap();
+				if let Err(e) = t2a.play_packet((con_id, from), packet) {
+					debug!(logger, "Failed to play packet"; "error" => %e);
+				}
+			}
+			Ok(())
+		});
+
+		// Wait for ctrl + c
+		tokio::select! {
+			send_audio = recv.recv() => {
+				if let Some(packet) = send_audio {
+					con.send_audio(packet)?;
+				} else {
+					info!(logger, "Audio sending stream was canceled");
+					break;
+				}
+			}
+			_ = tokio::signal::ctrl_c() => { break; }
+			r = events => {
+				r?;
+				bail!("Disconnected");
+			}
+		};
+	}
+
+	// Disconnect
+	con.disconnect(DisconnectOptions::new())?;
+	con.events().for_each(|_| future::ready(())).await;
+    Ok(())
+}

src/ts_voice/audio_to_ts.rs

@@ -0,0 +1,182 @@
+use std::sync::{Arc, Mutex};
+
+use anyhow::{format_err, Result};
+use audiopus::coder::Encoder;
+use futures::prelude::*;
+use sdl2::audio::{AudioCallback, AudioDevice, AudioSpec, AudioSpecDesired, AudioStatus};
+use sdl2::AudioSubsystem;
+use slog::{debug, error, o, Logger};
+use tokio::sync::mpsc;
+use tokio::task::LocalSet;
+use tokio::time::{self, Duration};
+use tokio_stream::wrappers::IntervalStream;
+use tsproto_packets::packets::{AudioData, CodecType, OutAudio, OutPacket};
+
+use super::*;
+
+pub struct AudioToTs {
+	logger: Logger,
+	audio_subsystem: AudioSubsystem,
+	listener: Arc<Mutex<Option<mpsc::Sender<OutPacket>>>>,
+	device: AudioDevice<SdlCallback>,
+
+	is_playing: bool,
+	volume: Arc<Mutex<f32>>,
+}
+
+struct SdlCallback {
+	logger: Logger,
+	spec: AudioSpec,
+	encoder: Encoder,
+	listener: Arc<Mutex<Option<mpsc::Sender<OutPacket>>>>,
+	volume: Arc<Mutex<f32>>,
+
+	opus_output: [u8; MAX_OPUS_FRAME_SIZE],
+}
+
+impl AudioToTs {
+	pub fn new(
+		logger: Logger, audio_subsystem: AudioSubsystem, local_set: &LocalSet,
+	) -> Result<Arc<Mutex<Self>>> {
+		let logger = logger.new(o!("pipeline" => "audio-to-ts"));
+		let listener = Arc::new(Mutex::new(Default::default()));
+		let volume = Arc::new(Mutex::new(1.0));
+
+		let device =
+			Self::open_capture(logger.clone(), &audio_subsystem, listener.clone(), volume.clone())?;
+
+		let res = Arc::new(Mutex::new(Self {
+			logger,
+			audio_subsystem,
+			listener,
+			device,
+
+			is_playing: false,
+			volume,
+		}));
+
+		Self::start(res.clone(), local_set);
+
+		Ok(res)
+	}
+
+	fn open_capture(
+		logger: Logger, audio_subsystem: &AudioSubsystem,
+		listener: Arc<Mutex<Option<mpsc::Sender<OutPacket>>>>, volume: Arc<Mutex<f32>>,
+	) -> Result<AudioDevice<SdlCallback>> {
+		let desired_spec = AudioSpecDesired {
+			freq: Some(48000),
+			channels: Some(1),
+			// Default sample size, 20 ms per packet
+			samples: Some(48000 / 50),
+		};
+
+		audio_subsystem.open_capture(None, &desired_spec, |spec| {
+			// This spec will always be the desired spec, the sdl wrapper passes
+			// zero as `allowed_changes`.
+			debug!(logger, "Got capture spec"; "spec" => ?spec, "driver" => audio_subsystem.current_audio_driver());
+			let opus_channels = if spec.channels == 1 {
+				audiopus::Channels::Mono
+			} else {
+				audiopus::Channels::Stereo
+			};
+
+			let encoder = Encoder::new(audiopus::SampleRate::Hz48000,
+				opus_channels, audiopus::Application::Voip)
+				.expect("Could not create encoder");
+
+			SdlCallback {
+				logger,
+				spec,
+				encoder,
+				listener,
+				volume,
+
+				opus_output: [0; MAX_OPUS_FRAME_SIZE],
+			}
+		}).map_err(|e| format_err!("SDL error: {}", e))
+	}
+
+	pub fn set_listener(&self, sender: mpsc::Sender<OutPacket>) {
+		let mut listener = self.listener.lock().unwrap();
+		*listener = Some(sender);
+	}
+
+	pub fn set_volume(&mut self, volume: f32) { *self.volume.lock().unwrap() = volume; }
+
+	pub fn set_playing(&mut self, playing: bool) {
+		if playing {
+			self.device.resume();
+		} else {
+			self.device.pause();
+		}
+		self.is_playing = playing;
+	}
+
+	fn start(a2t: Arc<Mutex<Self>>, local_set: &LocalSet) {
+		local_set.spawn_local(
+			IntervalStream::new(time::interval(Duration::from_secs(1))).for_each(move |_| {
+				let mut a2t = a2t.lock().unwrap();
+				if a2t.device.status() == AudioStatus::Stopped {
+					// Try to reconnect to audio
+					match Self::open_capture(
+						a2t.logger.clone(),
+						&a2t.audio_subsystem,
+						a2t.listener.clone(),
+						a2t.volume.clone(),
+					) {
+						Ok(d) => {
+							a2t.device = d;
+							debug!(a2t.logger, "Reconnected to capture device");
+							if a2t.is_playing {
+								a2t.device.resume();
+							}
+						}
+						Err(e) => {
+							error!(a2t.logger, "Failed to open capture device"; "error" => %e);
+						}
+					};
+				}
+				future::ready(())
+			}),
+		);
+	}
+}
+
+impl AudioCallback for SdlCallback {
+	type Channel = f32;
+	fn callback(&mut self, buffer: &mut [Self::Channel]) {
+		// Handle volume
+		let volume = *self.volume.lock().unwrap();
+		if volume != 1.0 {
+			for d in &mut *buffer {
+				*d *= volume;
+			}
+		}
+
+		match self.encoder.encode_float(buffer, &mut self.opus_output[..]) {
+			Err(e) => {
+				error!(self.logger, "Failed to encode opus"; "error" => %e);
+			}
+			Ok(len) => {
+				// Create packet
+				let codec = if self.spec.channels == 1 {
+					CodecType::OpusVoice
+				} else {
+					CodecType::OpusMusic
+				};
+				let packet =
+					OutAudio::new(&AudioData::C2S { id: 0, codec, data: &self.opus_output[..len] });
+
+				// Write into packet sink
+				let mut listener = self.listener.lock().unwrap();
+				if let Some(lis) = &mut *listener {
+					match lis.try_send(packet) {
+						Err(mpsc::error::TrySendError::Closed(_)) => *listener = None,
+						_ => {}
+					}
+				}
+			}
+		}
+	}
+}

src/ts_voice/mod.rs

@@ -0,0 +1,41 @@
+use std::sync::{Arc, Mutex};
+
+use anyhow::Result;
+use slog::Logger;
+use tokio::task::LocalSet;
+
+use audio_to_ts::AudioToTs;
+use ts_to_audio::TsToAudio;
+
+pub mod audio_to_ts;
+pub mod ts_to_audio;
+
+/// The usual frame size.
+///
+/// Use 48 kHz, 20 ms frames (50 per second) and mono data (1 channel).
+/// This means 1920 samples and 7.5 kiB.
+const USUAL_FRAME_SIZE: usize = 48000 / 50;
+
+/// The maximum size of an opus frame is 1275 as from RFC6716.
+const MAX_OPUS_FRAME_SIZE: usize = 1275;
+
+#[derive(Clone)]
+pub struct AudioData {
+	pub a2ts: Arc<Mutex<AudioToTs>>,
+	pub ts2a: Arc<Mutex<TsToAudio>>,
+}
+
+pub fn start(logger: Logger, local_set: &LocalSet) -> Result<AudioData> {
+	let sdl_context = sdl2::init().unwrap();
+
+	let audio_subsystem = sdl_context.audio().unwrap();
+	// SDL automatically disables the screensaver, enable it again
+	if let Ok(video_subsystem) = sdl_context.video() {
+		video_subsystem.enable_screen_saver();
+	}
+
+	let ts2a = TsToAudio::new(logger.clone(), audio_subsystem.clone(), local_set)?;
+	let a2ts = AudioToTs::new(logger.clone(), audio_subsystem, local_set)?;
+
+	Ok(AudioData { a2ts, ts2a })
+}

src/ts_voice/ts_to_audio.rs

@@ -0,0 +1,124 @@
+use std::sync::{Arc, Mutex};
+
+use anyhow::{format_err, Result};
+use futures::prelude::*;
+use sdl2::audio::{AudioCallback, AudioDevice, AudioSpecDesired, AudioStatus};
+use sdl2::AudioSubsystem;
+use slog::{debug, error, o, Logger};
+use tokio::task::LocalSet;
+use tokio::time::{self, Duration};
+use tokio_stream::wrappers::IntervalStream;
+use tsclientlib::ClientId;
+use tsproto_packets::packets::InAudioBuf;
+
+use super::*;
+use crate::ConnectionId;
+
+type Id = (ConnectionId, ClientId);
+type AudioHandler = tsclientlib::audio::AudioHandler<Id>;
+
+pub struct TsToAudio {
+	logger: Logger,
+	audio_subsystem: AudioSubsystem,
+	device: AudioDevice<SdlCallback>,
+	data: Arc<Mutex<AudioHandler>>,
+}
+
+struct SdlCallback {
+	data: Arc<Mutex<AudioHandler>>,
+}
+
+impl TsToAudio {
+	pub fn new(
+		logger: Logger, audio_subsystem: AudioSubsystem, local_set: &LocalSet,
+	) -> Result<Arc<Mutex<Self>>> {
+		let logger = logger.new(o!("pipeline" => "ts-to-audio"));
+		let data = Arc::new(Mutex::new(AudioHandler::new(logger.clone())));
+
+		let device = Self::open_playback(logger.clone(), &audio_subsystem, data.clone())?;
+
+		let res = Arc::new(Mutex::new(Self { logger, audio_subsystem, device, data }));
+
+		Self::start(res.clone(), local_set);
+
+		Ok(res)
+	}
+
+	fn open_playback(
+		logger: Logger, audio_subsystem: &AudioSubsystem, data: Arc<Mutex<AudioHandler>>,
+	) -> Result<AudioDevice<SdlCallback>> {
+		let desired_spec = AudioSpecDesired {
+			freq: Some(48000),
+			channels: Some(2),
+			samples: Some(USUAL_FRAME_SIZE as u16),
+		};
+
+		audio_subsystem.open_playback(None, &desired_spec, move |spec| {
+			// This spec will always be the desired spec, the sdl wrapper passes
+			// zero as `allowed_changes`.
+			debug!(logger, "Got playback spec"; "spec" => ?spec, "driver" => audio_subsystem.current_audio_driver());
+			SdlCallback {
+				data,
+			}
+		}).map_err(|e| format_err!("SDL error: {}", e))
+	}
+
+	fn start(t2a: Arc<Mutex<Self>>, local_set: &LocalSet) {
+		local_set.spawn_local(
+			IntervalStream::new(time::interval(Duration::from_secs(1))).for_each(move |_| {
+				let mut t2a = t2a.lock().unwrap();
+
+				if t2a.device.status() == AudioStatus::Stopped {
+					// Try to reconnect to audio
+					match Self::open_playback(
+						t2a.logger.clone(),
+						&t2a.audio_subsystem,
+						t2a.data.clone(),
+					) {
+						Ok(d) => {
+							t2a.device = d;
+							debug!(t2a.logger, "Reconnected to playback device");
+						}
+						Err(e) => {
+							error!(t2a.logger, "Failed to open playback device"; "error" => %e);
+						}
+					};
+				}
+
+				let data_empty = t2a.data.lock().unwrap().get_queues().is_empty();
+				if t2a.device.status() == AudioStatus::Paused && !data_empty {
+					debug!(t2a.logger, "Resuming playback");
+					t2a.device.resume();
+				} else if t2a.device.status() == AudioStatus::Playing && data_empty {
+					debug!(t2a.logger, "Pausing playback");
+					t2a.device.pause();
+				}
+				future::ready(())
+			}),
+		);
+	}
+
+	pub(crate) fn play_packet(&mut self, id: Id, packet: InAudioBuf) -> Result<()> {
+		let mut data = self.data.lock().unwrap();
+		data.handle_packet(id, packet)?;
+
+		if self.device.status() == AudioStatus::Paused {
+			debug!(self.logger, "Resuming playback");
+			self.device.resume();
+		}
+		Ok(())
+	}
+}
+
+impl AudioCallback for SdlCallback {
+	type Channel = f32;
+	fn callback(&mut self, buffer: &mut [Self::Channel]) {
+		// Clear buffer
+		for d in &mut *buffer {
+			*d = 0.0;
+		}
+
+		let mut data = self.data.lock().unwrap();
+		data.fill_buffer(buffer);
+	}
+}