From 03dab6d8299ccad819c6df622d67c916bc79c613 Mon Sep 17 00:00:00 2001
From: Andrei Zbikowski <b1naryth1ef@users.noreply.github.com>
Date: Mon, 10 Apr 2017 22:23:09 -0700
Subject: [PATCH] Voice Send Support (#17)

* First pass at voice sending

* more voice

* Refactor playables a bit, general fixes n stuff

* Cleanup

* Voice encryption, dep version bump, etc fixes

* Remove debugging, don't open a pipe for stderr

* Refactor playables

This is still a very lose concept, need to think about what the actual
differences between encoders and playables are. Also some rough edges in
general with the frame/sample calculations.

However, this still feels miles ahead of the previous iteration.

* Properly reset state when resuming from a pause

* rework playables/encoding/etc a bit

* Add a proxy, allow for more pipin'

* Cleanup, etc

* Fix resuming from a pause lerping music timestamp

* Fix some incorrect bounds checks, add MemoryBufferedPlayable
---
 .gitignore              |   1 +
 disco/bot/bot.py        |   6 +-
 disco/gateway/events.py |   2 +
 disco/voice/__init__.py |   3 +
 disco/voice/client.py   | 135 +++++++++++-----
 disco/voice/opus.py     | 149 +++++++++++++++++
 disco/voice/playable.py | 347 ++++++++++++++++++++++++++++++++++++++++
 disco/voice/player.py   | 122 ++++++++++++++
 examples/music.py       |  52 ++++++
 requirements.txt        |   7 +-
 10 files changed, 775 insertions(+), 49 deletions(-)
 create mode 100644 disco/voice/opus.py
 create mode 100644 disco/voice/playable.py
 create mode 100644 disco/voice/player.py
 create mode 100644 examples/music.py

diff --git a/.gitignore b/.gitignore
index 54cb9741..87a6c02a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,4 @@ dist/
 disco*.egg-info/
 docs/_build
 storage.db
+*.dca
diff --git a/disco/bot/bot.py b/disco/bot/bot.py
index 7693f507..8ad0fca1 100644
--- a/disco/bot/bot.py
+++ b/disco/bot/bot.py
@@ -355,10 +355,10 @@ def on_message_create(self, event):
         if event.message.author.id == self.client.state.me.id:
             return
 
-        if self.config.commands_allow_edit:
-            self.last_message_cache[event.message.channel_id] = (event.message, False)
+        result = self.handle_message(event.message)
 
-        self.handle_message(event.message)
+        if self.config.commands_allow_edit:
+            self.last_message_cache[event.message.channel_id] = (event.message, result)
 
     def on_message_update(self, event):
         if self.config.commands_allow_edit:
diff --git a/disco/gateway/events.py b/disco/gateway/events.py
index 5033b031..9d9d323e 100644
--- a/disco/gateway/events.py
+++ b/disco/gateway/events.py
@@ -49,6 +49,8 @@ def create(cls, obj, client):
         """
         Create this GatewayEvent class from data and the client.
         """
+        cls.raw_data = obj
+
         # If this event is wrapping a model, pull its fields
         if hasattr(cls, '_wraps_model'):
             alias, model = cls._wraps_model
diff --git a/disco/voice/__init__.py b/disco/voice/__init__.py
index e69de29b..b4a7f6c5 100644
--- a/disco/voice/__init__.py
+++ b/disco/voice/__init__.py
@@ -0,0 +1,3 @@
+from disco.voice.client import *
+from disco.voice.player import *
+from disco.voice.playable import *
diff --git a/disco/voice/client.py b/disco/voice/client.py
index 69fe9d51..d0d3e092 100644
--- a/disco/voice/client.py
+++ b/disco/voice/client.py
@@ -3,6 +3,8 @@
 import struct
 import time
 
+import nacl.secret
+
 from holster.enum import Enum
 from holster.emitter import Emitter
 
@@ -22,11 +24,6 @@
     VOICE_CONNECTED=6,
 )
 
-# TODO:
-#   - player implementation
-#   - encryption
-#   - cleanup
-
 
 class VoiceException(Exception):
     def __init__(self, msg, client):
@@ -38,12 +35,40 @@ class UDPVoiceClient(LoggingClass):
     def __init__(self, vc):
         super(UDPVoiceClient, self).__init__()
         self.vc = vc
+
+        # The underlying UDP socket
         self.conn = None
+
+        # Connection information
         self.ip = None
         self.port = None
+
         self.run_task = None
         self.connected = False
 
+    def send_frame(self, frame, sequence=None, timestamp=None):
+        # Convert the frame to a bytearray
+        frame = bytearray(frame)
+
+        # First, pack the header (TODO: reuse bytearray?)
+        header = bytearray(24)
+        header[0] = 0x80
+        header[1] = 0x78
+        struct.pack_into('>H', header, 2, sequence or self.vc.sequence)
+        struct.pack_into('>I', header, 4, timestamp or self.vc.timestamp)
+        struct.pack_into('>i', header, 8, self.vc.ssrc)
+
+        # Now encrypt the payload with the nonce as a header
+        raw = self.vc.secret_box.encrypt(bytes(frame), bytes(header)).ciphertext
+
+        # Send the header (sans nonce padding) plus the payload
+        self.send(header[:12] + raw)
+
+        # Increment our sequence counter
+        self.vc.sequence += 1
+        if self.vc.sequence >= 65535:
+            self.vc.sequence = 0
+
     def run(self):
         while True:
             self.conn.recvfrom(4096)
@@ -54,26 +79,29 @@ def send(self, data):
     def disconnect(self):
         self.run_task.kill()
 
-    def connect(self, host, port, timeout=10):
+    def connect(self, host, port, timeout=10, addrinfo=None):
         self.ip = socket.gethostbyname(host)
         self.port = port
 
         self.conn = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 
-        # Send discovery packet
-        packet = bytearray(70)
-        struct.pack_into('>I', packet, 0, self.vc.ssrc)
-        self.send(packet)
+        if addrinfo:
+            ip, port = addrinfo
+        else:
+            # Send discovery packet
+            packet = bytearray(70)
+            struct.pack_into('>I', packet, 0, self.vc.ssrc)
+            self.send(packet)
 
-        # Wait for a response
-        try:
-            data, addr = gevent.spawn(lambda: self.conn.recvfrom(70)).get(timeout=timeout)
-        except gevent.Timeout:
-            return (None, None)
+            # Wait for a response
+            try:
+                data, addr = gevent.spawn(lambda: self.conn.recvfrom(70)).get(timeout=timeout)
+            except gevent.Timeout:
+                return (None, None)
 
-        # Read IP and port
-        ip = str(data[4:]).split('\x00', 1)[0]
-        port = struct.unpack('<H', data[-2:])[0]
+            # Read IP and port
+            ip = str(data[4:]).split('\x00', 1)[0]
+            port = struct.unpack('<H', data[-2:])[0]
 
         # Spawn read thread so we don't max buffers
         self.connected = True
@@ -86,30 +114,45 @@ class VoiceClient(LoggingClass):
     def __init__(self, channel, encoder=None):
         super(VoiceClient, self).__init__()
 
-        assert channel.is_voice, 'Cannot spawn a VoiceClient for a non-voice channel'
+        if not channel.is_voice:
+            raise ValueError('Cannot spawn a VoiceClient for a non-voice channel')
+
         self.channel = channel
         self.client = self.channel.client
         self.encoder = encoder or JSONEncoder
 
+        # Bind to some WS packets
         self.packets = Emitter(gevent.spawn)
         self.packets.on(VoiceOPCode.READY, self.on_voice_ready)
         self.packets.on(VoiceOPCode.SESSION_DESCRIPTION, self.on_voice_sdp)
 
-        # State
+        # State + state change emitter
         self.state = VoiceState.DISCONNECTED
-        self.connected = gevent.event.Event()
+        self.state_emitter = Emitter(gevent.spawn)
+
+        # Connection metadata
         self.token = None
         self.endpoint = None
         self.ssrc = None
         self.port = None
+        self.secret_box = None
         self.udp = None
 
+        # Voice data state
+        self.sequence = 0
+        self.timestamp = 0
+
         self.update_listener = None
 
         # Websocket connection
         self.ws = None
         self.heartbeat_task = None
 
+    def set_state(self, state):
+        prev_state = self.state
+        self.state = state
+        self.state_emitter.emit(state, prev_state)
+
     def heartbeat(self, interval):
         while True:
             self.send(VoiceOPCode.HEARTBEAT, time.time() * 1000)
@@ -128,7 +171,7 @@ def send(self, op, data):
         }), self.encoder.OPCODE)
 
     def on_voice_ready(self, data):
-        self.state = VoiceState.CONNECTING
+        self.set_state(VoiceState.CONNECTING)
         self.ssrc = data['ssrc']
         self.port = data['port']
 
@@ -146,18 +189,20 @@ def on_voice_ready(self, data):
             'data': {
                 'port': port,
                 'address': ip,
-                'mode': 'plain'
+                'mode': 'xsalsa20_poly1305'
             }
         })
 
-    def on_voice_sdp(self, _):
+    def on_voice_sdp(self, sdp):
+        # Create a secret box for encryption/decryption
+        self.secret_box = nacl.secret.SecretBox(bytes(bytearray(sdp['secret_key'])))
+
         # Toggle speaking state so clients learn of our SSRC
         self.set_speaking(True)
         self.set_speaking(False)
         gevent.sleep(0.25)
 
-        self.state = VoiceState.CONNECTED
-        self.connected.set()
+        self.set_state(VoiceState.CONNECTED)
 
     def on_voice_server_update(self, data):
         if self.channel.guild_id != data.guild_id or not data.token:
@@ -167,30 +212,28 @@ def on_voice_server_update(self, data):
             return
 
         self.token = data.token
-        self.state = VoiceState.AUTHENTICATING
+        self.set_state(VoiceState.AUTHENTICATING)
 
         self.endpoint = data.endpoint.split(':', 1)[0]
-        self.ws = Websocket(
-            'wss://' + self.endpoint,
-            on_message=self.on_message,
-            on_error=self.on_error,
-            on_open=self.on_open,
-            on_close=self.on_close,
-        )
+        self.ws = Websocket('wss://' + self.endpoint)
+        self.ws.emitter.on('on_open', self.on_open)
+        self.ws.emitter.on('on_error', self.on_error)
+        self.ws.emitter.on('on_close', self.on_close)
+        self.ws.emitter.on('on_message', self.on_message)
         self.ws.run_forever()
 
-    def on_message(self, _, msg):
+    def on_message(self, msg):
         try:
             data = self.encoder.decode(msg)
             self.packets.emit(VoiceOPCode[data['op']], data['d'])
         except:
             self.log.exception('Failed to parse voice gateway message: ')
 
-    def on_error(self, _, err):
-        # TODO
+    def on_error(self, err):
+        # TODO: raise an exception here
         self.log.warning('Voice websocket error: {}'.format(err))
 
-    def on_open(self, _):
+    def on_open(self):
         self.send(VoiceOPCode.IDENTIFY, {
             'server_id': self.channel.guild_id,
             'user_id': self.client.state.me.id,
@@ -198,12 +241,15 @@ def on_open(self, _):
             'token': self.token
         })
 
-    def on_close(self, _, code, error):
-        # TODO
+    def on_close(self, code, error):
         self.log.warning('Voice websocket disconnected (%s, %s)', code, error)
 
+        if self.state == VoiceState.CONNECTED:
+            self.log.info('Attempting voice reconnection')
+            self.connect()
+
     def connect(self, timeout=5, mute=False, deaf=False):
-        self.state = VoiceState.AWAITING_ENDPOINT
+        self.set_state(VoiceState.AWAITING_ENDPOINT)
 
         self.update_listener = self.client.events.on('VoiceServerUpdate', self.on_voice_server_update)
 
@@ -214,11 +260,11 @@ def connect(self, timeout=5, mute=False, deaf=False):
             'channel_id': int(self.channel.id),
         })
 
-        if not self.connected.wait(timeout) or self.state != VoiceState.CONNECTED:
+        if not self.state_emitter.once(VoiceState.CONNECTED, timeout=timeout):
             raise VoiceException('Failed to connect to voice', self)
 
     def disconnect(self):
-        self.state = VoiceState.DISCONNECTED
+        self.set_state(VoiceState.DISCONNECTED)
 
         if self.heartbeat_task:
             self.heartbeat_task.kill()
@@ -236,3 +282,6 @@ def disconnect(self):
             'guild_id': int(self.channel.guild_id),
             'channel_id': None,
         })
+
+    def send_frame(self, *args, **kwargs):
+        self.udp.send_frame(*args, **kwargs)
diff --git a/disco/voice/opus.py b/disco/voice/opus.py
new file mode 100644
index 00000000..246dbed0
--- /dev/null
+++ b/disco/voice/opus.py
@@ -0,0 +1,149 @@
+import sys
+import array
+import ctypes
+import ctypes.util
+
+from holster.enum import Enum
+
+from disco.util.logging import LoggingClass
+
+
+c_int_ptr = ctypes.POINTER(ctypes.c_int)
+c_int16_ptr = ctypes.POINTER(ctypes.c_int16)
+c_float_ptr = ctypes.POINTER(ctypes.c_float)
+
+
+class EncoderStruct(ctypes.Structure):
+    pass
+
+
+class DecoderStruct(ctypes.Structure):
+    pass
+
+
+EncoderStructPtr = ctypes.POINTER(EncoderStruct)
+DecoderStructPtr = ctypes.POINTER(DecoderStruct)
+
+
+class BaseOpus(LoggingClass):
+    BASE_EXPORTED = {
+        'opus_strerror': ([ctypes.c_int], ctypes.c_char_p),
+    }
+
+    EXPORTED = {}
+
+    def __init__(self, library_path=None):
+        self.path = library_path or self.find_library()
+        self.lib = ctypes.cdll.LoadLibrary(self.path)
+
+        methods = {}
+        methods.update(self.BASE_EXPORTED)
+        methods.update(self.EXPORTED)
+
+        for name, item in methods.items():
+            func = getattr(self.lib, name)
+
+            if item[0]:
+                func.argtypes = item[0]
+
+            func.restype = item[1]
+
+            setattr(self, name, func)
+
+    @staticmethod
+    def find_library():
+        if sys.platform == 'win32':
+            raise Exception('Cannot auto-load opus on Windows, please specify full library path')
+
+        return ctypes.util.find_library('opus')
+
+
+Application = Enum(
+    AUDIO=2049,
+    VOIP=2048,
+    LOWDELAY=2051
+)
+
+
+Control = Enum(
+    SET_BITRATE=4002,
+    SET_BANDWIDTH=4008,
+    SET_FEC=4012,
+    SET_PLP=4014,
+)
+
+
+class OpusEncoder(BaseOpus):
+    EXPORTED = {
+        'opus_encoder_get_size': ([ctypes.c_int], ctypes.c_int),
+        'opus_encoder_create': ([ctypes.c_int, ctypes.c_int, ctypes.c_int, c_int_ptr], EncoderStructPtr),
+        'opus_encode': ([EncoderStructPtr, c_int16_ptr, ctypes.c_int, ctypes.c_char_p, ctypes.c_int32], ctypes.c_int32),
+        'opus_encoder_ctl': (None, ctypes.c_int32),
+        'opus_encoder_destroy': ([EncoderStructPtr], None),
+    }
+
+    def __init__(self, sampling_rate, channels, application=Application.AUDIO, library_path=None):
+        super(OpusEncoder, self).__init__(library_path)
+        self.sampling_rate = sampling_rate
+        self.channels = channels
+        self.application = application
+
+        self._inst = None
+
+    @property
+    def inst(self):
+        if not self._inst:
+            self._inst = self.create()
+            self.set_bitrate(128)
+            self.set_fec(True)
+            self.set_expected_packet_loss_percent(0.15)
+        return self._inst
+
+    def set_bitrate(self, kbps):
+        kbps = min(128, max(16, int(kbps)))
+        ret = self.opus_encoder_ctl(self.inst, int(Control.SET_BITRATE), kbps * 1024)
+
+        if ret < 0:
+            raise Exception('Failed to set bitrate to {}: {}'.format(kbps, ret))
+
+    def set_fec(self, value):
+        ret = self.opus_encoder_ctl(self.inst, int(Control.SET_FEC), int(value))
+
+        if ret < 0:
+            raise Exception('Failed to set FEC to {}: {}'.format(value, ret))
+
+    def set_expected_packet_loss_percent(self, perc):
+        ret = self.opus_encoder_ctl(self.inst, int(Control.SET_PLP), min(100, max(0, int(perc * 100))))
+
+        if ret < 0:
+            raise Exception('Failed to set PLP to {}: {}'.format(perc, ret))
+
+    def create(self):
+        ret = ctypes.c_int()
+        result = self.opus_encoder_create(self.sampling_rate, self.channels, self.application.value, ctypes.byref(ret))
+
+        if ret.value != 0:
+            raise Exception('Failed to create opus encoder: {}'.format(ret.value))
+
+        return result
+
+    def __del__(self):
+        if hasattr(self, '_inst') and self._inst:
+            self.opus_encoder_destroy(self._inst)
+            self._inst = None
+
+    def encode(self, pcm, frame_size):
+        max_data_bytes = len(pcm)
+        pcm = ctypes.cast(pcm, c_int16_ptr)
+        data = (ctypes.c_char * max_data_bytes)()
+
+        ret = self.opus_encode(self.inst, pcm, frame_size, data, max_data_bytes)
+        if ret < 0:
+            raise Exception('Failed to encode: {}'.format(ret))
+
+        # TODO: py3
+        return array.array('b', data[:ret]).tostring()
+
+
+class OpusDecoder(BaseOpus):
+    pass
diff --git a/disco/voice/playable.py b/disco/voice/playable.py
new file mode 100644
index 00000000..aaeb7bd7
--- /dev/null
+++ b/disco/voice/playable.py
@@ -0,0 +1,347 @@
+import abc
+import six
+import types
+import gevent
+import struct
+import subprocess
+
+
+from gevent.queue import Queue
+
+from disco.voice.opus import OpusEncoder
+
+
+try:
+    from cStringIO import cStringIO as StringIO
+except:
+    from StringIO import StringIO
+
+
+OPUS_HEADER_SIZE = struct.calcsize('<h')
+
+
+# Play from file:
+#  OpusFilePlayable(open('myfile.opus', 'r'))
+#  PCMFileInput(open('myfile.pcm', 'r')).pipe(DCADOpusEncoder) => OpusPlayable
+#  FFMpegInput.youtube_dl('youtube.com/yolo').pipe(DCADOpusEncoder) => OpusPlayable
+#  FFMpegInput.youtube_dl('youtube.com/yolo').pipe(OpusEncoder).pipe(DuplexStream, open('cache_file.opus', 'w')) => OpusPlayable
+
+
+class AbstractOpus(object):
+    def __init__(self, sampling_rate=48000, frame_length=20, channels=2):
+        self.sampling_rate = sampling_rate
+        self.frame_length = frame_length
+        self.channels = 2
+        self.sample_size = 2 * self.channels
+        self.samples_per_frame = int(self.sampling_rate / 1000 * self.frame_length)
+        self.frame_size = self.samples_per_frame * self.sample_size
+
+
+class BaseUtil(object):
+    def pipe(self, other, *args, **kwargs):
+        child = other(self, *args, **kwargs)
+        setattr(child, 'metadata', self.metadata)
+        setattr(child, '_parent', self)
+        return child
+
+    @property
+    def metadata(self):
+        return self._metadata
+
+    @metadata.setter
+    def metadata(self, value):
+        self._metadata = value
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BasePlayable(BaseUtil):
+    @abc.abstractmethod
+    def next_frame(self):
+        raise NotImplementedError
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseInput(BaseUtil):
+    @abc.abstractmethod
+    def read(self, size):
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def fileobj(self):
+        raise NotImplementedError
+
+
+class OpusFilePlayable(BasePlayable, AbstractOpus):
+    """
+    An input which reads opus data from a file or file-like object.
+    """
+    def __init__(self, fobj, *args, **kwargs):
+        super(OpusFilePlayable, self).__init__(*args, **kwargs)
+        self.fobj = fobj
+        self.done = False
+
+    def next_frame(self):
+        if self.done:
+            return None
+
+        header = self.fobj.read(OPUS_HEADER_SIZE)
+        if len(header) < OPUS_HEADER_SIZE:
+            self.done = True
+            return None
+
+        data_size = struct.unpack('<h', header)[0]
+        data = self.fobj.read(data_size)
+        if len(data) < data_size:
+            self.done = True
+            return None
+
+        return data
+
+
+class FFmpegInput(BaseInput, AbstractOpus):
+    def __init__(self, source='-', command='avconv', streaming=False, **kwargs):
+        super(FFmpegInput, self).__init__(**kwargs)
+        if source:
+            self.source = source
+        self.streaming = streaming
+        self.command = command
+
+        self._buffer = None
+        self._proc = None
+
+    def read(self, sz):
+        if self.streaming:
+            raise TypeError('Cannot read from a streaming FFmpegInput')
+
+        # First read blocks until the subprocess finishes
+        if not self._buffer:
+            data, _ = self.proc.communicate()
+            self._buffer = StringIO(data)
+
+        # Subsequent reads can just do dis thang
+        return self._buffer.read(sz)
+
+    def fileobj(self):
+        if self.streaming:
+            return self.proc.stdout
+        else:
+            return self
+
+    @property
+    def proc(self):
+        if not self._proc:
+            if callable(self.source):
+                self.source = self.source(self)
+
+            if isinstance(self.source, (tuple, list)):
+                self.source, self.metadata = self.source
+
+            args = [
+                self.command,
+                '-i', str(self.source),
+                '-f', 's16le',
+                '-ar', str(self.sampling_rate),
+                '-ac', str(self.channels),
+                '-loglevel', 'warning',
+                'pipe:1'
+            ]
+            self._proc = subprocess.Popen(args, stdin=None, stdout=subprocess.PIPE)
+        return self._proc
+
+
+class YoutubeDLInput(FFmpegInput):
+    def __init__(self, url=None, ie_info=None, *args, **kwargs):
+        super(YoutubeDLInput, self).__init__(None, *args, **kwargs)
+        self._url = url
+        self._ie_info = ie_info
+        self._info = None
+
+    @property
+    def info(self):
+        if not self._info:
+            import youtube_dl
+            ydl = youtube_dl.YoutubeDL({'format': 'webm[abr>0]/bestaudio/best'})
+
+            if self._url:
+                obj = ydl.extract_info(self._url, download=False, process=False)
+                if 'entries' in obj:
+                    self._ie_info = obj['entries']
+                else:
+                    self._ie_info = [obj]
+
+            self._info = ydl.process_ie_result(self._ie_info, download=False)
+        return self._info
+
+    @property
+    def _metadata(self):
+        return self.info
+
+    @classmethod
+    def many(cls, url, *args, **kwargs):
+        import youtube_dl
+
+        ydl = youtube_dl.YoutubeDL({'format': 'webm[abr>0]/bestaudio/best'})
+        info = ydl.extract_info(url, download=False, process=False)
+
+        if 'entries' not in info:
+            yield cls(ie_info=info, *args, **kwargs)
+            raise StopIteration
+
+        for item in info['entries']:
+            yield cls(ie_info=item, *args, **kwargs)
+
+    @property
+    def source(self):
+        return self.info['url']
+
+
+class BufferedOpusEncoderPlayable(BasePlayable, AbstractOpus, OpusEncoder):
+    def __init__(self, source, *args, **kwargs):
+        self.source = source
+        self.frames = Queue(kwargs.pop('queue_size', 4096))
+        super(BufferedOpusEncoderPlayable, self).__init__(*args, **kwargs)
+        gevent.spawn(self._encoder_loop)
+
+    def _encoder_loop(self):
+        while self.source:
+            raw = self.source.read(self.frame_size)
+            if len(raw) < self.frame_size:
+                break
+
+            self.frames.put(self.encode(raw, self.samples_per_frame))
+            gevent.idle()
+        self.source = None
+
+    def next_frame(self):
+        if not self.source:
+            return None
+        return self.frames.get()
+
+
+class DCADOpusEncoderPlayable(BasePlayable, AbstractOpus, OpusEncoder):
+    def __init__(self, source, *args, **kwargs):
+        self.source = source
+        self.command = kwargs.pop('command', 'dcad')
+        super(DCADOpusEncoderPlayable, self).__init__(*args, **kwargs)
+
+        self._done = False
+        self._proc = None
+
+    @property
+    def proc(self):
+        if not self._proc:
+            source = obj = self.source.fileobj()
+            if not hasattr(obj, 'fileno'):
+                source = subprocess.PIPE
+
+            self._proc = subprocess.Popen([
+                self.command,
+                '--channels', str(self.channels),
+                '--rate', str(self.sampling_rate),
+                '--size', str(self.samples_per_frame),
+                '--bitrate', '128',
+                '--fec',
+                '--packet-loss-percent', '30',
+                '--input', 'pipe:0',
+                '--output', 'pipe:1',
+            ], stdin=source, stdout=subprocess.PIPE)
+
+            def writer():
+                while True:
+                    data = obj.read(2048)
+                    if len(data) > 0:
+                        self._proc.stdin.write(data)
+                    if len(data) < 2048:
+                        break
+
+            if source == subprocess.PIPE:
+                gevent.spawn(writer)
+        return self._proc
+
+    def next_frame(self):
+        if self._done:
+            return None
+
+        header = self.proc.stdout.read(OPUS_HEADER_SIZE)
+        if len(header) < OPUS_HEADER_SIZE:
+            self._done = True
+            return
+
+        size = struct.unpack('<h', header)[0]
+
+        data = self.proc.stdout.read(size)
+        if len(data) < size:
+            self._done = True
+            return
+
+        return data
+
+
+class FileProxyPlayable(BasePlayable, AbstractOpus):
+    def __init__(self, other, output, *args, **kwargs):
+        self.flush = kwargs.pop('flush', False)
+        super(FileProxyPlayable, self).__init__(*args, **kwargs)
+        self.other = other
+        self.output = output
+
+    def next_frame(self):
+        frame = self.other.next_frame()
+
+        if frame:
+            self.output.write(struct.pack('<h', len(frame)))
+            self.output.write(frame)
+
+            if self.flush:
+                self.output.flush()
+        else:
+            self.output.flush()
+            self.output.close()
+        return frame
+
+
+class PlaylistPlayable(BasePlayable, AbstractOpus):
+    def __init__(self, items, *args, **kwargs):
+        super(PlaylistPlayable, self).__init__(*args, **kwargs)
+        self.items = items
+        self.now_playing = None
+
+    def _get_next(self):
+        if isinstance(self.items, types.GeneratorType):
+            return next(self.items, None)
+        return self.items.pop()
+
+    def next_frame(self):
+        if not self.items:
+            return
+
+        if not self.now_playing:
+            self.now_playing = self._get_next()
+            if not self.now_playing:
+                return
+
+        frame = self.now_playing.next_frame()
+        if not frame:
+            return self.next_frame()
+
+        return frame
+
+
+class MemoryBufferedPlayable(BasePlayable, AbstractOpus):
+    def __init__(self, other, *args, **kwargs):
+        from gevent.queue import Queue
+
+        super(MemoryBufferedPlayable, self).__init__(*args, **kwargs)
+        self.frames = Queue()
+        self.other = other
+        gevent.spawn(self._buffer)
+
+    def _buffer(self):
+        while True:
+            frame = self.other.next_frame()
+            if not frame:
+                break
+            self.frames.put(frame)
+        self.frames.put(None)
+
+    def next_frame(self):
+        return self.frames.get()
diff --git a/disco/voice/player.py b/disco/voice/player.py
new file mode 100644
index 00000000..9abe0759
--- /dev/null
+++ b/disco/voice/player.py
@@ -0,0 +1,122 @@
+import time
+import gevent
+
+from six.moves import queue
+from holster.enum import Enum
+from holster.emitter import Emitter
+
+from disco.voice.client import VoiceState
+
+
+class Player(object):
+    Events = Enum(
+        'START_PLAY',
+        'STOP_PLAY',
+        'PAUSE_PLAY',
+        'RESUME_PLAY',
+        'DISCONNECT'
+    )
+
+    def __init__(self, client):
+        self.client = client
+
+        # Queue contains playable items
+        self.queue = queue.Queue()
+
+        # Whether we're playing music (true for lifetime)
+        self.playing = True
+
+        # Set to an event when playback is paused
+        self.paused = None
+
+        # Current playing item
+        self.now_playing = None
+
+        # Current play task
+        self.play_task = None
+
+        # Core task
+        self.run_task = gevent.spawn(self.run)
+
+        # Event triggered when playback is complete
+        self.complete = gevent.event.Event()
+
+        # Event emitter for metadata
+        self.events = Emitter(gevent.spawn)
+
+    def disconnect(self):
+        self.client.disconnect()
+        self.events.emit(self.Events.DISCONNECT)
+
+    def skip(self):
+        self.play_task.kill()
+
+    def pause(self):
+        if self.paused:
+            return
+        self.paused = gevent.event.Event()
+        self.events.emit(self.Events.PAUSE_PLAY)
+
+    def resume(self):
+        self.paused.set()
+        self.paused = None
+        self.events.emit(self.Events.RESUME_PLAY)
+
+    def play(self, item):
+        # Grab the first frame before we start anything else, sometimes playables
+        #  can do some lengthy async tasks here to setup the playable and we
+        #  don't want that lerp the first N frames of the playable into playing
+        #  faster
+        frame = item.next_frame()
+        if frame is None:
+            return
+
+        start = time.time()
+        loops = 0
+
+        while True:
+            loops += 1
+
+            if self.paused:
+                self.client.set_speaking(False)
+                self.paused.wait()
+                gevent.sleep(2)
+                self.client.set_speaking(True)
+                start = time.time()
+                loops = 0
+
+            if self.client.state == VoiceState.DISCONNECTED:
+                return
+
+            if self.client.state != VoiceState.CONNECTED:
+                self.client.state_emitter.wait(VoiceState.CONNECTED)
+
+            self.client.send_frame(frame)
+            self.client.timestamp += item.samples_per_frame
+
+            frame = item.next_frame()
+            if frame is None:
+                return
+
+            next_time = start + 0.02 * loops
+            delay = max(0, 0.02 + (next_time - time.time()))
+            gevent.sleep(delay)
+
+    def run(self):
+        self.client.set_speaking(True)
+
+        while self.playing:
+            self.now_playing = self.queue.get()
+
+            self.events.emit(self.Events.START_PLAY, self.now_playing)
+            self.play_task = gevent.spawn(self.play, self.now_playing)
+            self.play_task.join()
+            self.events.emit(self.Events.STOP_PLAY, self.now_playing)
+
+            if self.client.state == VoiceState.DISCONNECTED:
+                self.playing = False
+                self.complete.set()
+                return
+
+        self.client.set_speaking(False)
+        self.disconnect()
diff --git a/examples/music.py b/examples/music.py
new file mode 100644
index 00000000..63232b9e
--- /dev/null
+++ b/examples/music.py
@@ -0,0 +1,52 @@
+from disco.bot import Plugin
+from disco.bot.command import CommandError
+from disco.voice.player import Player
+from disco.voice.playable import FFmpegInput, DCADOpusEncoderPlayable
+from disco.voice.client import VoiceException
+
+
+class MusicPlugin(Plugin):
+    def load(self, ctx):
+        super(MusicPlugin, self).load(ctx)
+        self.guilds = {}
+
+    @Plugin.command('join')
+    def on_join(self, event):
+        if event.guild.id in self.guilds:
+            return event.msg.reply("I'm already playing music here.")
+
+        state = event.guild.get_member(event.author).get_voice_state()
+        if not state:
+            return event.msg.reply('You must be connected to voice to use that command.')
+
+        try:
+            client = state.channel.connect()
+        except VoiceException as e:
+            return event.msg.reply('Failed to connect to voice: `{}`'.format(e))
+
+        self.guilds[event.guild.id] = Player(client)
+        self.guilds[event.guild.id].complete.wait()
+        del self.guilds[event.guild.id]
+
+    def get_player(self, guild_id):
+        if guild_id not in self.guilds:
+            raise CommandError("I'm not currently playing music here.")
+        return self.guilds.get(guild_id)
+
+    @Plugin.command('leave')
+    def on_leave(self, event):
+        player = self.get_player(event.guild.id)
+        player.disconnect()
+
+    @Plugin.command('play', '<url:str>')
+    def on_play(self, event, url):
+        item = FFmpegInput.youtube_dl(url).pipe(DCADOpusEncoderPlayable)
+        self.get_player(event.guild.id).queue.put(item)
+
+    @Plugin.command('pause')
+    def on_pause(self, event):
+        self.get_player(event.guild.id).pause()
+
+    @Plugin.command('resume')
+    def on_resume(self, event):
+        self.get_player(event.guild.id).resume()
diff --git a/requirements.txt b/requirements.txt
index 3b0894b0..ed9a25c2 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,6 +1,7 @@
-gevent==1.1.2
+gevent==1.2.1
 holster==1.0.11
 inflection==0.3.1
-requests==2.11.1
+requests==2.13.0
 six==1.10.0
-websocket-client==0.37.0
+websocket-client==0.40.0
+pynacl==1.1.2