Repository: discordjs/voice
Pull request #118 (2021-06-12 15:25:35)

feat(Util): add Opus audio probing utility
**Please describe the changes this PR makes and why it should be merged:**

This adds an Opus audio probing utility that will help with deciding the best way to handle an audio stream. It can determine whether the input stream is a valid WebM or Ogg Opus stream for playing in Discord voice channels (i.e. an Opus stream at 48kHz sample rate and 2 audio channels). This allows you to demux the stream, rather than blindly passing it to FFmpeg, for improved performance. For example:

```ts
import { createReadStream } from 'fs';
import { StreamType, demuxProbe, createAudioResource } from '@discordjs/voice';

async function start() {
  // Example verifications
  assert.equal(await demuxProbe(createReadStream('file.mp3')), StreamType.Arbitrary);
  assert.equal(await demuxProbe(createReadStream('file.opus')), StreamType.OggOpus);
  assert.equal(await demuxProbe(createReadStream('file.webm')), StreamType.WebmOpus);

  // Example of creating an audio resource using the probe
  const stream = createReadStream('./song.unknownextension');
  createAudioResource(stream, { inputType: await demuxProbe(stream) });
}
```

The motivation for this PR was that a lot of users were having issues with `ytdl-core` abort errors, e.g. see #116. The `demuxProbe` util allows us to use `youtube-dl` instead without sacrificing the same performance benefits. The music bot example has been updated to reflect this.

To-do:
- [x] Tests

**Status and versioning classification:**

- Code changes have been tested against the Discord API, or there are no code changes
- This PR changes the library's interface (methods or parameters added)

codecov[bot]: # [Codecov](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs) Report
> Merging [#118](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs) (5e4f798) into [main](https://codecov.io/gh/discordjs/voice/commit/2442d617dc7ed5d6530e0127be4184c5aa052580?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs) (2442d61) will **decrease** coverage by `3.11%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/discordjs/voice/pull/118/graphs/tree.svg?width=650&height=150&src=pr&token=u7oQ23UoxX&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs)](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs)

```diff
@@            Coverage Diff             @@
##             main     #118      +/-   ##
==========================================
- Coverage   63.55%   60.44%   -3.12%     
==========================================
  Files          18       19       +1     
  Lines         815      857      +42     
  Branches      195      204       +9     
==========================================
  Hits          518      518              
- Misses        295      337      +42     
  Partials        2        2              
```


| [Impacted Files](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs) | Coverage Δ | |
|---|---|---|
| [src/util/demuxProbe.ts](https://codecov.io/gh/discordjs/voice/pull/118/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs#diff-c3JjL3V0aWwvZGVtdXhQcm9iZS50cw==) | `0.00% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs). Last update [2442d61...5e4f798](https://codecov.io/gh/discordjs/voice/pull/118?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=discordjs).


Patch:
diff --git a/examples/music-bot/README.md b/examples/music-bot/README.md
index afb110c..22be695 100644
--- a/examples/music-bot/README.md
+++ b/examples/music-bot/README.md
@@ -9,8 +9,9 @@ If you're looking to make your own music bot that is fairly simple, this example
 ## Usage
 
 ```bash
-# Clone the main repository
+# Clone the main repository, and then run:
 $ npm install
+$ npm run build
 
 # Open this example and install dependencies
 $ cd examples/music-bot
diff --git a/examples/music-bot/package.json b/examples/music-bot/package.json
index 2b9bb57..40d6316 100644
--- a/examples/music-bot/package.json
+++ b/examples/music-bot/package.json
@@ -16,11 +16,13 @@
 	"dependencies": {
 		"@discordjs/opus": "^0.5.0",
 		"discord-api-types": "^0.18.1",
-		"discord.js": "^13.0.0-dev.c6aeebb18d6b969f7c8bdb1b719883d4384dd03e",
+		"discord.js": "^13.0.0-dev.02693bc02f45980d8165820a103220f0027b96b7",
 		"libsodium-wrappers": "^0.7.9",
-		"ytdl-core-discord": "^1.3.0"
+		"youtube-dl-exec": "^1.2.4",
+		"ytdl-core": "^4.8.2"
 	},
 	"devDependencies": {
-		"tsconfig-paths": "^3.9.0"
+		"tsconfig-paths": "^3.9.0",
+		"typescript": "~4.2.2"
 	}
 }
diff --git a/examples/music-bot/src/music/track.ts b/examples/music-bot/src/music/track.ts
index d77694e..15241c1 100644
--- a/examples/music-bot/src/music/track.ts
+++ b/examples/music-bot/src/music/track.ts
@@ -1,5 +1,6 @@
-import ytdl, { getInfo } from 'ytdl-core-discord';
-import { AudioResource, createAudioResource, StreamType } from '@discordjs/voice';
+import { getInfo } from 'ytdl-core';
+import { AudioResource, createAudioResource, demuxProbe } from '@discordjs/voice';
+import { raw as ytdl } from 'youtube-dl-exec';
 
 /**
  * This is the data required to create a Track object
@@ -42,8 +43,36 @@ export class Track implements TrackData {
 	/**
 	 * Creates an AudioResource from this Track.
 	 */
-	public async createAudioResource(): Promise<AudioResource<Track>> {
-		return createAudioResource(await ytdl(this.url), { metadata: this, inputType: StreamType.Opus });
+	public createAudioResource(): Promise<AudioResource<Track>> {
+		return new Promise((resolve, reject) => {
+			const process = ytdl(
+				this.url,
+				{
+					o: '-',
+					q: true,
+					f: 'bestaudio[ext=webm+acodec=opus+asr=48000]/bestaudio',
+					r: '100K',
+				},
+				{ stdio: ['ignore', 'pipe', 'ignore'] },
+			);
+			if (!process.stdout) {
+				reject(new Error('No stdout'));
+				return;
+			}
+			const stream = process.stdout;
+			const onError = (error: Error) => {
+				if (!process.killed) process.kill();
+				stream.resume();
+				reject(error);
+			};
+			process
+				.once('spawn', () => {
+					demuxProbe(stream)
+						.then((probe) => resolve(createAudioResource(probe.stream, { metadata: this, inputType: probe.type })))
+						.catch(onError);
+				})
+				.catch(onError);
+		});
 	}
 
 	/**
diff --git a/examples/music-bot/tsconfig.json b/examples/music-bot/tsconfig.json
index 7ebef24..9778261 100644
--- a/examples/music-bot/tsconfig.json
+++ b/examples/music-bot/tsconfig.json
@@ -6,8 +6,9 @@
 		"rootDir": "src",
 		"paths": {
 			"@discordjs/voice": ["../../"],
-			"sodium": ["./node_modules/sodium"],
 			"@discordjs/opus": ["./node_modules/@discordjs/opus"],
+			"sodium": ["./node_modules/sodium"],
+			"libsodium-wrappers": ["./node_modules/libsodium-wrappers"],
 			"tweetnacl": ["./node_modules/tweetnacl"]
 		}
 	},
diff --git a/package-lock.json b/package-lock.json
index 8e635a5..1afb935 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -11,7 +11,7 @@
 			"dependencies": {
 				"@types/ws": "^7.4.4",
 				"discord-api-types": "^0.18.1",
-				"prism-media": "^1.2.9",
+				"prism-media": "^1.3.1",
 				"tiny-typed-emitter": "^2.0.3",
 				"ws": "^7.4.4"
 			},
@@ -8940,9 +8940,9 @@
 			}
 		},
 		"node_modules/prism-media": {
-			"version": "1.2.9",
-			"resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.2.9.tgz",
-			"integrity": "sha512-UHCYuqHipbTR1ZsXr5eg4JUmHER8Ss4YEb9Azn+9zzJ7/jlTtD1h0lc4g6tNx3eMlB8Mp6bfll0LPMAV4R6r3Q==",
+			"version": "1.3.1",
+			"resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.3.1.tgz",
+			"integrity": "sha512-nyYAa3KB4qteJIqdguKmwxTJgy55xxUtkJ3uRnOvO5jO+frci+9zpRXw6QZVcfDeva3S654fU9+26P2OSTzjHw==",
 			"peerDependencies": {
 				"@discordjs/opus": "^0.5.0",
 				"ffmpeg-static": "^4.2.7 || ^3.0.0 || ^2.4.0",
@@ -17427,9 +17427,9 @@
 			}
 		},
 		"prism-media": {
-			"version": "1.2.9",
-			"resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.2.9.tgz",
-			"integrity": "sha512-UHCYuqHipbTR1ZsXr5eg4JUmHER8Ss4YEb9Azn+9zzJ7/jlTtD1h0lc4g6tNx3eMlB8Mp6bfll0LPMAV4R6r3Q=="
+			"version": "1.3.1",
+			"resolved": "https://registry.npmjs.org/prism-media/-/prism-media-1.3.1.tgz",
+			"integrity": "sha512-nyYAa3KB4qteJIqdguKmwxTJgy55xxUtkJ3uRnOvO5jO+frci+9zpRXw6QZVcfDeva3S654fU9+26P2OSTzjHw=="
 		},
 		"progress": {
 			"version": "2.0.3",
diff --git a/package.json b/package.json
index fb6943f..1b5b287 100644
--- a/package.json
+++ b/package.json
@@ -40,7 +40,7 @@
 	"dependencies": {
 		"@types/ws": "^7.4.4",
 		"discord-api-types": "^0.18.1",
-		"prism-media": "^1.2.9",
+		"prism-media": "^1.3.1",
 		"tiny-typed-emitter": "^2.0.3",
 		"ws": "^7.4.4"
 	},
diff --git a/src/util/demuxProbe.ts b/src/util/demuxProbe.ts
new file mode 100644
index 0000000..3e100b4
--- /dev/null
+++ b/src/util/demuxProbe.ts
@@ -0,0 +1,113 @@
+import { Readable } from 'stream';
+import { opus } from 'prism-media';
+import { noop } from './util';
+import { StreamType } from '..';
+
+/**
+ * Takes an Opus Head, and verifies whether the associated Opus audio is suitable to play in a Discord voice channel.
+ * @param opusHead The Opus Head to validate
+ * @returns true if suitable to play in a Discord voice channel, false otherwise
+ */
+export function validateDiscordOpusHead(opusHead: Buffer): boolean {
+	const channels = opusHead.readUInt8(9);
+	const sampleRate = opusHead.readUInt32LE(12);
+	return channels === 2 && sampleRate === 48000;
+}
+
+/**
+ * The resulting information after probing an audio stream
+ */
+export interface ProbeInfo {
+	/**
+	 * The readable audio stream to use. You should use this rather than the input stream, as the probing
+	 * function can sometimes read the input stream to its end and cause the stream to close.
+	 */
+	stream: Readable;
+	/**
+	 * The recommended stream type for this audio stream
+	 */
+	type: StreamType;
+}
+
+/**
+ * Attempt to probe a readable stream to figure out whether it can be demuxed using an Ogg or WebM Opus demuxer.
+ * @param stream The readable stream to probe
+ * @param probeSize The number of bytes to attempt to read before giving up on the probe
+ * @param validator The Opus Head validator function
+ * @experimental
+ */
+export function demuxProbe(
+	stream: Readable,
+	probeSize = 1024,
+	validator = validateDiscordOpusHead,
+): Promise<ProbeInfo> {
+	return new Promise((resolve, reject) => {
+		// Preconditions
+		if (stream.readableObjectMode) reject(new Error('Cannot probe a readable stream in object mode'));
+		if (stream.readableEnded) reject(new Error('Cannot probe a stream that has ended'));
+
+		let readBuffer = Buffer.alloc(0);
+
+		let resolved: StreamType | undefined = undefined;
+
+		const finish = (type: StreamType) => {
+			stream.off('data', onData);
+			stream.off('close', onClose);
+			stream.off('end', onClose);
+			stream.pause();
+			resolved = type;
+			if (stream.readableEnded) {
+				resolve({
+					stream: Readable.from(readBuffer),
+					type,
+				});
+			} else {
+				if (readBuffer.length > 0) {
+					stream.push(readBuffer);
+				}
+				resolve({
+					stream,
+					type,
+				});
+			}
+		};
+
+		const foundHead = (type: StreamType) => (head: Buffer) => {
+			if (validator(head)) {
+				finish(type);
+			}
+		};
+
+		const webm = new opus.WebmDemuxer();
+		webm.once('error', noop);
+		webm.on('head', foundHead(StreamType.WebmOpus));
+
+		const ogg = new opus.OggDemuxer();
+		ogg.once('error', noop);
+		ogg.on('head', foundHead(StreamType.OggOpus));
+
+		const onClose = () => {
+			if (!resolved) {
+				finish(StreamType.Arbitrary);
+			}
+		};
+
+		const onData = (buffer: Buffer) => {
+			readBuffer = Buffer.concat([readBuffer, buffer]);
+
+			webm.write(buffer);
+			ogg.write(buffer);
+
+			if (readBuffer.length >= probeSize) {
+				stream.off('data', onData);
+				stream.pause();
+				process.nextTick(onClose);
+			}
+		};
+
+		stream.once('error', reject);
+		stream.on('data', onData);
+		stream.once('close', onClose);
+		stream.once('end', onClose);
+	});
+}
diff --git a/src/util/index.ts b/src/util/index.ts
index b7294df..9b9efe7 100644
--- a/src/util/index.ts
+++ b/src/util/index.ts
@@ -1,3 +1,4 @@
 export * from './generateDependencyReport';
 export * from './entersState';
 export * from './adapter';
+export * from './demuxProbe';


Test patch:
diff --git a/src/util/__tests__/demuxProbe.test.ts b/src/util/__tests__/demuxProbe.test.ts
new file mode 100644
index 0000000..c12ddb8
--- /dev/null
+++ b/src/util/__tests__/demuxProbe.test.ts
@@ -0,0 +1,119 @@
+import { demuxProbe } from '../demuxProbe';
+import { opus as _opus } from 'prism-media';
+import { Readable } from 'stream';
+import { StreamType } from '../../audio';
+import EventEmitter, { once } from 'events';
+
+jest.mock('prism-media');
+
+const WebmDemuxer = _opus.WebmDemuxer as unknown as jest.Mock<_opus.WebmDemuxer>;
+const OggDemuxer = _opus.OggDemuxer as unknown as jest.Mock<_opus.OggDemuxer>;
+
+async function* gen(n: number) {
+	for (let i = 0; i < n; i++) {
+		yield Buffer.from([i]);
+		await nextTick();
+	}
+}
+
+function range(n: number) {
+	return Buffer.from(Array.from(Array(n).keys()));
+}
+
+const validHead = Buffer.from([
+	0x4f, 0x70, 0x75, 0x73, 0x48, 0x65, 0x61, 0x64, 0x01, 0x02, 0x38, 0x01, 0x80, 0xbb, 0, 0, 0, 0, 0,
+]);
+
+const invalidHead = Buffer.from([
+	0x4f, 0x70, 0x75, 0x73, 0x48, 0x65, 0x61, 0x64, 0x01, 0x01, 0x38, 0x01, 0x80, 0xbb, 0, 0, 0, 0, 0,
+]);
+
+async function collectStream(stream: Readable): Promise<Buffer> {
+	let output = Buffer.alloc(0);
+	await once(stream, 'readable');
+	for await (const data of stream) {
+		output = Buffer.concat([output, data]);
+	}
+	return output;
+}
+
+function nextTick() {
+	return new Promise((resolve) => process.nextTick(resolve));
+}
+
+describe('demuxProbe', () => {
+	const webmWrite: jest.Mock<(buffer: Buffer) => void> = jest.fn();
+	const oggWrite: jest.Mock<(buffer: Buffer) => void> = jest.fn();
+
+	beforeAll(() => {
+		WebmDemuxer.prototype = {
+			...WebmDemuxer,
+			...EventEmitter.prototype,
+			write: webmWrite,
+		};
+		OggDemuxer.prototype = {
+			...OggDemuxer,
+			...EventEmitter.prototype,
+			write: oggWrite,
+		};
+	});
+
+	beforeEach(() => {
+		webmWrite.mockReset();
+		oggWrite.mockReset();
+	});
+
+	test('Defaults to arbitrary', async () => {
+		const stream = Readable.from(gen(10), { objectMode: false });
+		const probe = await demuxProbe(stream);
+		expect(probe.type).toBe(StreamType.Arbitrary);
+		await expect(collectStream(probe.stream)).resolves.toEqual(range(10));
+	});
+
+	test('Detects WebM', async () => {
+		const stream = Readable.from(gen(10), { objectMode: false });
+		webmWrite.mockImplementation(function mock(data: Buffer) {
+			if (data[0] === 5) this.emit('head', validHead);
+		} as any);
+		const probe = await demuxProbe(stream);
+		expect(probe.type).toBe(StreamType.WebmOpus);
+		await expect(collectStream(probe.stream)).resolves.toEqual(range(10));
+	});
+
+	test('Detects Ogg', async () => {
+		const stream = Readable.from(gen(10), { objectMode: false });
+		oggWrite.mockImplementation(function mock(data: Buffer) {
+			if (data[0] === 5) this.emit('head', validHead);
+		} as any);
+		const probe = await demuxProbe(stream);
+		expect(probe.type).toBe(StreamType.OggOpus);
+		await expect(collectStream(probe.stream)).resolves.toEqual(range(10));
+	});
+
+	test('Rejects invalid OpusHead', async () => {
+		const stream = Readable.from(gen(10), { objectMode: false });
+		oggWrite.mockImplementation(function mock(data: Buffer) {
+			if (data[0] === 5) this.emit('head', invalidHead);
+		} as any);
+		const probe = await demuxProbe(stream);
+		expect(probe.type).toBe(StreamType.Arbitrary);
+		await expect(collectStream(probe.stream)).resolves.toEqual(range(10));
+	});
+
+	test('Gives up on larger streams', async () => {
+		const stream = Readable.from(gen(8192), { objectMode: false });
+		const probe = await demuxProbe(stream);
+		expect(probe.type).toBe(StreamType.Arbitrary);
+		await expect(collectStream(probe.stream)).resolves.toEqual(range(8192));
+	});
+
+	test('Propagates errors', async () => {
+		const testError = new Error('test error');
+		const stream = new Readable({
+			read() {
+				this.destroy(testError);
+			},
+		});
+		await expect(demuxProbe(stream)).rejects.toBe(testError);
+	});
+});