fix(Voice*): fix speaking event and voice receive (#3749)

* fix(Voice*): synthesize speaking event from UDP packets

* fix(VoiceReceiver): skip over undocumented Discord byte

See #3555

* fix(VoiceConnection): play frame silence before emitting ready

* typings: account for changes in private api

src/client/voice/VoiceConnection.js

@@ -4,9 +4,14 @@ const Util = require('../../util/Util');
 const Constants = require('../../util/Constants');
 const AudioPlayer = require('./player/AudioPlayer');
 const VoiceReceiver = require('./receiver/VoiceReceiver');
+const SingleSilence = require('./util/SingleSilence');
 const EventEmitter = require('events').EventEmitter;
 const Prism = require('prism-media');
 
+// The delay between packets when a user is considered to have stopped speaking
+// https://github.com/discordjs/discord.js/issues/3524#issuecomment-540373200
+const DISCORD_SPEAKING_DELAY = 250;
+
 /**
  * Represents a connection to a guild's voice server.
  * ```js
@@ -101,12 +106,19 @@ class VoiceConnection extends EventEmitter {
     });
 
     /**
-     * Map SSRC to speaking values
-     * @type {Map<number, boolean>}
+     * Map SSRC to user id
+     * @type {Map<number, Snowflake>}
      * @private
      */
     this.ssrcMap = new Map();
 
+    /**
+     * Map user id to speaking timeout
+     * @type {Map<Snowflake, Timeout>}
+     * @private
+     */
+    this.speakingTimeouts = new Map();
+
     /**
      * Object that wraps contains the `ws` and `udp` sockets of this voice connection
      * @type {Object}
@@ -342,7 +354,7 @@ class VoiceConnection extends EventEmitter {
       ws.removeAllListeners('error');
       ws.removeAllListeners('ready');
       ws.removeAllListeners('sessionDescription');
-      ws.removeAllListeners('speaking');
+      ws.removeAllListeners('startSpeaking');
       ws.shutdown();
     }
 
@@ -374,7 +386,7 @@ class VoiceConnection extends EventEmitter {
     udp.on('error', err => this.emit('error', err));
     ws.on('ready', this.onReady.bind(this));
     ws.on('sessionDescription', this.onSessionDescription.bind(this));
-    ws.on('speaking', this.onSpeaking.bind(this));
+    ws.on('startSpeaking', this.onStartSpeaking.bind(this));
   }
 
   /**
@@ -386,6 +398,7 @@ class VoiceConnection extends EventEmitter {
     this.authentication.port = port;
     this.authentication.ssrc = ssrc;
     this.sockets.udp.createUDPSocket(ip);
+    this.sockets.udp.socket.on('message', this.onUDPMessage.bind(this));
   }
 
   /**
@@ -399,12 +412,29 @@ class VoiceConnection extends EventEmitter {
     this.authentication.secretKey = secret;
 
     this.status = Constants.VoiceStatus.CONNECTED;
-    /**
-     * Emitted once the connection is ready, when a promise to join a voice channel resolves,
-     * the connection will already be ready.
-     * @event VoiceConnection#ready
-     */
-    this.emit('ready');
+    const ready = () => {
+      /**
+       * Emitted once the connection is ready, when a promise to join a voice channel resolves,
+       * the connection will already be ready.
+       * @event VoiceConnection#ready
+       */
+      this.emit('ready');
+    };
+    if (this.dispatcher) {
+      ready();
+    } else {
+      // This serves to provide support for voice receive, sending audio is required to receive it.
+      this.playOpusStream(new SingleSilence()).once('end', ready);
+    }
+  }
+
+  /**
+   * Invoked whenever a user initially starts speaking.
+   * @param {Object} data The speaking data
+   * @private
+   */
+  onStartSpeaking({ user_id, ssrc }) {
+    this.ssrcMap.set(+ssrc, user_id);
   }
 
   /**
@@ -412,10 +442,9 @@ class VoiceConnection extends EventEmitter {
    * @param {Object} data The received data
    * @private
    */
-  onSpeaking({ user_id, ssrc, speaking }) {
+  onSpeaking({ user_id, speaking }) {
     const guild = this.channel.guild;
     const user = this.client.users.get(user_id);
-    this.ssrcMap.set(+ssrc, user);
     if (!speaking) {
       for (const receiver of this.receivers) {
         receiver.stoppedSpeaking(user);
@@ -431,6 +460,35 @@ class VoiceConnection extends EventEmitter {
     guild._memberSpeakUpdate(user_id, speaking);
   }
 
+  /**
+   * Handles synthesizing of the speaking event.
+   * @param {Buffer} buffer Received packet from the UDP socket
+   * @private
+   */
+  onUDPMessage(buffer) {
+    const ssrc = +buffer.readUInt32BE(8).toString(10);
+    const user = this.client.users.get(this.ssrcMap.get(ssrc));
+    if (!user) return;
+
+    let speakingTimeout = this.speakingTimeouts.get(ssrc);
+    if (typeof speakingTimeout === 'undefined') {
+      this.onSpeaking({ user_id: user.id, ssrc, speaking: true });
+    } else {
+      this.client.clearTimeout(speakingTimeout);
+    }
+
+    speakingTimeout = this.client.setTimeout(() => {
+      try {
+        this.onSpeaking({ user_id: user.id, ssrc, speaking: false });
+        this.client.clearTimeout(speakingTimeout);
+        this.speakingTimeouts.delete(ssrc);
+      } catch (ex) {
+        // Connection already closed, ignore
+      }
+    }, DISCORD_SPEAKING_DELAY);
+    this.speakingTimeouts.set(ssrc, speakingTimeout);
+  }
+
   /**
    * Options that can be passed to stream-playing methods:
    * @typedef {Object} StreamOptions

src/client/voice/VoiceWebSocket.js

@@ -184,9 +184,9 @@ class VoiceWebSocket extends EventEmitter {
         /**
          * Emitted whenever a speaking packet is received.
          * @param {Object} data
-         * @event VoiceWebSocket#speaking
+         * @event VoiceWebSocket#startSpeaking
          */
-        this.emit('speaking', packet.d);
+        this.emit('startSpeaking', packet.d);
         break;
       default:
         /**

src/client/voice/receiver/VoiceReceiver.js

@@ -43,7 +43,7 @@ class VoiceReceiver extends EventEmitter {
 
     this._listener = msg => {
       const ssrc = +msg.readUInt32BE(8).toString(10);
-      const user = this.voiceConnection.ssrcMap.get(ssrc);
+      const user = connection.client.users.get(connection.ssrcMap.get(ssrc));
       if (!user) {
         if (!this.queues.has(ssrc)) this.queues.set(ssrc, []);
         this.queues.get(ssrc).push(msg);
@@ -175,9 +175,9 @@ class VoiceReceiver extends EventEmitter {
         }
         offset += 1 + (0b1111 & (byte >> 4));
       }
-      while (data[offset] === 0) {
-        offset++;
-      }
+      // Skip over undocumented Discord byte
+      offset++;
+
       data = data.slice(offset);
     }
 

src/client/voice/util/Silence.js

@@ -0,0 +1,16 @@
+const { Readable } = require('stream');
+
+const SILENCE_FRAME = Buffer.from([0xF8, 0xFF, 0xFE]);
+
+/**
+ * A readable emitting silent opus frames.
+ * @extends {Readable}
+ * @private
+ */
+class Silence extends Readable {
+  _read() {
+    this.push(SILENCE_FRAME);
+  }
+}
+
+module.exports = Silence;

src/client/voice/util/SingleSilence.js

@@ -0,0 +1,17 @@
+const Silence = require('./Silence');
+
+/**
+ * Only emits a single silent opus frame.
+ * This is used as a workaround for Discord now requiring
+ * silence to be sent before being able to receive audio.
+ * @extends {Silence}
+ * @private
+ */
+class SingleSilence extends Silence {
+  _read() {
+    super._read();
+    this.push(null);
+  }
+}
+
+module.exports = SingleSilence;

typings/index.d.ts

@@ -1415,7 +1415,8 @@ declare module 'discord.js' {
 		constructor(voiceManager: ClientVoiceManager, channel: VoiceChannel);
 		private authentication: object;
 		private sockets: object;
-		private ssrcMap: Map<number, boolean>;
+		private ssrcMap: Map<number, Snowflake>;
+		private speakingTimeouts: Map<number, NodeJS.Timer>;
 		private authenticate(): void;
 		private authenticateFailed(reason: string): void;
 		private checkAuthenticated(): void;
@@ -1540,14 +1541,14 @@ declare module 'discord.js' {
 
 		public on(event: 'ready', listener: (packet: object) => void): this;
 		public on(event: 'sessionDescription', listener: (encryptionMode: string, secretKey: SecretKey) => void): this;
-		public on(event: 'speaking', listener: (data: object) => void): this;
+		public on(event: 'startSpeaking', listener: (data: object) => void): this;
 		public on(event: 'unknownPacket', listener: (packet: object) => void): this;
 		public on(event: 'warn', listener: (warn: string) => void): this;
 		public on(event: string, listener: Function): this;
 
 		public once(event: 'ready', listener: (packet: object) => void): this;
 		public once(event: 'sessionDescription', listener: (encryptionMode: string, secretKey: SecretKey) => void): this;
-		public once(event: 'speaking', listener: (data: object) => void): this;
+		public once(event: 'startSpeaking', listener: (data: object) => void): this;
 		public once(event: 'unknownPacket', listener: (packet: object) => void): this;
 		public once(event: 'warn', listener: (warn: string) => void): this;
 		public once(event: string, listener: Function): this;