通过 socket.io 流式传输实时音频

Question

如何使用 socket.io 将实时音频从一个客户端流式传输到可能的多个客户端？

我已经到了可以在同一个标​​签中录制音频和播放音频的地步。

这是我目前的代码：

$(document).ready(function () {
    var socket = io("ws://127.0.0.1:4385");

    if (navigator.mediaDevices) {
        console.log('getUserMedia supported.');

        var constraints = { audio: true };

        navigator.mediaDevices.getUserMedia(constraints)
            .then(function (stream) {

                let ctx = new AudioContext();
                let source = ctx.createMediaStreamSource(stream);
                let destination = ctx.createMediaStreamDestination();
                source.connect(ctx.destination);
            })
            .catch(function (err) {
                console.log('The following error occurred: ' + err);
            })
    }
});

如何将该音频流发送到我的 socket.io 服务器，然后再发送回另一个客户端？

我听说过 WebRTC，但我不想要点对点解决方案，因为如果有多个客户端想要收听音频，这会给客户端带来负担。

必须有一种方法可以检索原始音频数据并将其发送到我的 socket.io 服务器，然后再将其发送回想要收听它的客户端。

Answer 1

经过多次反复试验，我找到了一个令我满意的解决方案。 这是客户端javascript。服务器端 socket.io 服务器只是将数据转发给正确的客户端，应该是微不足道的。

里面也有一些前端的东西。无视就好。

main.js

var socket;
var ctx;
var playbackBuffers = {};
var audioWorkletNodes = {};
var isMuted = true;

$(document).ready(function () {

    $('#login-form').on('submit', function (e) {
        e.preventDefault();
        $('#login-view').hide();
        $('#content-view').show();
        connectToVoiceServer($('#username').val());
        createAudioContext();

        $('#mute-toggle').click(function () {
            isMuted = !isMuted;
            if (isMuted) {
                $('#mute-toggle').html('<i class="bi bi-mic-mute"></i>');
            } else {
                $('#mute-toggle').html('<i class="bi bi-mic"></i>');
            }
        });

        if (navigator.mediaDevices) {
            setupRecordWorklet();
        } else {
            // TODO: Display warning can not access microphone
        }
    });
});

function setupRecordWorklet() {
    navigator.mediaDevices.getUserMedia({ audio: true })
        .then(async function (stream) {
            await ctx.audioWorklet.addModule('./js/record-processor.js');
            let src = ctx.createMediaStreamSource(stream);

            const processor = new AudioWorkletNode(ctx, 'record-processor');

            let recordBuffer;
            processor.port.onmessage = (e) => {
                if (e.data.eventType === 'buffer') {
                    recordBuffer = new Float32Array(e.data.buffer);
                }
                if (e.data.eventType === 'data' && !isMuted) {
                    socket.volatile.emit('voice', { id: socket.id, buffer: recordBuffer.slice(e.data.start, e.data.end).buffer });
                }
            }
            src.connect(processor);
        })
        .catch(function (err) {
            console.log('The following error occurred: ' + err);
        });

    socket.on('voice', data => {
        if (playbackBuffers[data.id]) {
            let buffer = new Float32Array(data.buffer);
            playbackBuffers[data.id].buffer.set(buffer, playbackBuffers[data.id].cursor);
            playbackBuffers[data.id].cursor += buffer.length;
            playbackBuffers[data.id].cursor %= buffer.length * 4;
        }
    });
}

function createAudioContext() {
    ctx = new AudioContext();
}

function connectToVoiceServer(username) {
    socket = io("wss://example.com", { query: `username=${username}` });

    socket.on("connect", function () {

    });

    socket.on('user:connect', function (user) {
        addUser(user.id, user.username);
    });

    socket.on('user:disconnect', function (id) {
        removeUser(id);
    });

    socket.on('user:list', function (users) {
        users.forEach(function (user) {
            addUser(user.id, user.username);
        });
    });
}

function addUser(id, username) {
    $('#user-list').append(`<li id="${id}" class="list-group-item text-truncate">${username}</li>`);
    addUserAudio(id);
}

function removeUser(id) {
    $('#' + id).remove();
    removeUserAudio(id);
}

async function addUserAudio(id) {
    await ctx.audioWorklet.addModule('./js/playback-processor.js');
    audioWorkletNodes[id] = new AudioWorkletNode(ctx, 'playback-processor');

    audioWorkletNodes[id].port.onmessage = (e) => {
        if (e.data.eventType === 'buffer') {
            playbackBuffers[id] = { cursor: 0, buffer: new Float32Array(e.data.buffer) };
        }
    }

    audioWorkletNodes[id].connect(ctx.destination);
}

function removeUserAudio(id) {
    audioWorkletNodes[id].disconnect();
    audioWorkletNodes[id] = undefined;
    playbackBuffers[id] = undefined;
}

record-processor.js

class RecordProcessor extends AudioWorkletProcessor {

    constructor() {
        super();
        this._cursor = 0;
        this._bufferSize = 8192 * 4;
        this._sharedBuffer = new SharedArrayBuffer(this._bufferSize);
        this._sharedView = new Float32Array(this._sharedBuffer);
        this.port.postMessage({
            eventType: 'buffer',
            buffer: this._sharedBuffer
        });
    }

    process(inputs, outputs) {

        for (let i = 0; i < inputs[0][0].length; i++) {
            this._sharedView[(i + this._cursor) % this._sharedView.length] = inputs[0][0][i];
        }

        if (((this._cursor + inputs[0][0].length) % (this._sharedView.length / 4)) === 0) {
            this.port.postMessage({
                eventType: 'data',
                start: this._cursor - this._sharedView.length / 4 + inputs[0][0].length,
                end: this._cursor + inputs[0][0].length
            });
        }

        this._cursor += inputs[0][0].length;
        this._cursor %= this._sharedView.length;

        return true;
    }
}

registerProcessor('record-processor', RecordProcessor);

playback-processor.js

class PlaybackProcessor extends AudioWorkletProcessor {

    constructor() {
        super();
        this._cursor = 0;
        this._bufferSize = 8192 * 4;
        this._sharedBuffer = new SharedArrayBuffer(this._bufferSize);
        this._sharedView = new Float32Array(this._sharedBuffer);
        this.port.postMessage({
            eventType: 'buffer',
            buffer: this._sharedBuffer
        });
    }

    process(inputs, outputs) {

        for (let i = 0; i < outputs[0][0].length; i++) {
            outputs[0][0][i] = this._sharedView[i + this._cursor];
            this._sharedView[i + this._cursor] = 0;
        }

        this._cursor += outputs[0][0].length;
        this._cursor %= this._sharedView.length;

        return true;
    }
}

registerProcessor('playback-processor', PlaybackProcessor);

注意事项：

1. 我正在使用 SharedArrayBuffers 读取/写入 AudioWorklet。为了使它们正常工作，您的服务器必须提供带有标题的网页：Cross-Origin-Opener-Policy=same-origin 和 Cross-Origin-Embedder-Policy=require-corp
2. 这将传输未压缩的非交错 IEEE754 32 位线性 PCM 音频。因此通过网络传输的数据将是巨大的。必须添加压缩！
3. 假设发送方和接收方的采样率相同