【流媒体服务器Mediasoup】 NodeJs与C++信令通信详解及Linux下管道通信的详解（五）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h> 
#include <sys/socket.h>
const char* str = "SOCKET PAIR TEST.";
int main(int argc, char* argv[]){	
	int    socket_pair[2];
	pid_t  id;
	//创建socketpair
	if(socketpair(AF_UNIX, SOCK_STREAM, 0, socket_pair) == -1){		
		printf("Error, socketpair create failed, errno(%d): %sn", errno, strerror(errno));
        return EXIT_FAILURE;
	}	
	//创建子进程
	id =fork();	
	if(id == 0){
		//子进程
		char buffer[512]={0, };
		//子进程关闭管道4
		close(socket_pair[1]);
		while(1){
			//发送数据给父进程
			printf("childer send n");
			write(socket_pair[0], str , strlen(str));
			sleep(1);			
			//接收父进程数据
			ssize_t len = read(socket_pair[0], buffer, sizeof(buffer));
			if(len > 0 ){
				buffer[len] = '';
				printf("childer: recv from parent : %s n",buffer);	
			}
		}		
	}else if(id > 0){
		//父进程
		char buffer[512]={0, };
		//父进程关闭管道3
		close(socket_pair[0]);		
		while(1){
			
			//接收子进程数据
			ssize_t len = read(socket_pair[1], buffer, sizeof(buffer));
			if(len > 0 ){
				buffer[len] = '';
				printf("father: recv from childer : %s n",buffer);				
			}
			sleep(1);
			//发送数据给父进程
			printf("father send n");
			write(socket_pair[1], str , strlen(str));
		}		
	}else{		
		printf("Error, fork failed, errno(%d): %sn", errno, strerror(errno));
        return EXIT_FAILURE;
	}	
	return 0;	
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
 * Launch as many mediasoup Workers as given in the configuration file.
 */
async function runMediasoupWorkers()
{
	const { numWorkers } = config.mediasoup;

	logger.info('running %d mediasoup Workers...', numWorkers);

	for (let i = 0; i < numWorkers; ++i)
	{     //服务启动后开始创建worker
		const worker = await mediasoup.createWorker(
			{
				logLevel   : config.mediasoup.workerSettings.logLevel,
				logTags    : config.mediasoup.workerSettings.logTags,
				rtcMinPort : Number(config.mediasoup.workerSettings.rtcMinPort),
				rtcMaxPort : Number(config.mediasoup.workerSettings.rtcMaxPort)
			});

        ....省略
    }
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class Worker extends EnhancedEventEmitter_1.EnhancedEventEmitter {
    /**
     * @private
     * @emits died - (error: Error)
     * @emits @success
     * @emits @failure - (error: Error)
     */
    constructor({ logLevel, logTags, rtcMinPort, rtcMaxPort, dtlsCertificateFile, dtlsPrivateKeyFile, appData }) {
        super();
        ...省略
        //启动核心文件 实际上新建woker即线程
        this._child = child_process_1.spawn(
        // command
        spawnBin, 
        // args
        spawnArgs,   
        // options
        {
            env: {
                MEDIASOUP_VERSION: '3.4.11'
            },
            detached: false,
            // fd 0 (stdin)   : Just ignore it.
            // fd 1 (stdout)  : Pipe it for 3rd libraries that log their own stuff.
            // fd 2 (stderr)  : Same as stdout.
            // fd 3 (channel) : Producer Channel fd.
            // fd 4 (channel) : Consumer Channel fd.
            stdio: ['ignore', 'pipe', 'pipe', 'pipe', 'pipe']
        });
        this._pid = this._child.pid;
        // 创建的管道交给channel处理
        this._channel = new Channel_1.Channel({
            producerSocket: this._child.stdio[3],
            consumerSocket: this._child.stdio[4],
            pid: this._pid
        });
        this._appData = appData;
        let spawnDone = false;
        // Listen for 'ready' notification.
        this._channel.once(String(this._pid), (event) => {
             ...省略
        });
        this._child.on('exit', (code, signal) => {
            ...省略
        });
        this._child.on('error', (error) => {
            ...省略
        });
        // Be ready for 3rd party worker libraries logging to stdout.
        this._child.stdout.on('data', (buffer) => {
            ...省略
        });
        // In case of a worker bug, mediasoup will log to stderr.
        this._child.stderr.on('data', (buffer) => {
            ...省略
        });
    }

1
2
3
4
5
this._channel = new Channel_1.Channel({
            producerSocket: this._child.stdio[3],
            consumerSocket: this._child.stdio[4],
            pid: this._pid
        });

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
request(method, internal, data) {
        return __awaiter(this, void 0, void 0, function* () {
                  ...省略
            //将数据写入管道 这里真正的是进行管道写入
            this._producerSocket.write(ns);
            //用异步方式去监听底层是否接受并处理信息，这里的确认结果和接收中的逻辑相匹配
            //会通过this._sents.set(id, sent); sent的里的resolve 或者 pReject 返回
            return new Promise((pResolve, pReject) => {
                const timeout = 1000 * (15 + (0.1 * this._sents.size));
                const sent = {
                    id: id,
                    method: method,
                    resolve: (data2) => {
                        if (!this._sents.delete(id))
                            return;
                        clearTimeout(sent.timer);
                        pResolve(data2);
                    },
                    reject: (error) => {
                        if (!this._sents.delete(id))
                            return;
                        clearTimeout(sent.timer);
                        pReject(error);
                    },
                    timer: setTimeout(() => {
                        if (!this._sents.delete(id))
                            return;
                        pReject(new Error('Channel request timeout'));
                    }, timeout),
                    close: () => {
                        clearTimeout(sent.timer);
                        pReject(new errors_1.InvalidStateError('Channel closed'));
                    }
                };
                // Add sent stuff to the map.
                this._sents.set(id, sent);
            });
        });
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class Channel extends EnhancedEventEmitter_1.EnhancedEventEmitter {

    constructor({ producerSocket, consumerSocket, pid }) {
        ...省略
        this._producerSocket = producerSocket;
        this._consumerSocket = consumerSocket;
        // Read Channel responses/notifications from the worker.
        //用于接受底层C++发来的信令数据
        this._consumerSocket.on('data', (buffer) => {
                ...省略
                try {
                    // We can receive JSON messages (Channel messages) or log strings.
                    switch (nsPayload[0]) {
                        // 123 = '{' (a Channel JSON messsage).
                        case 123:
                            //真正的处理有效的信令数据
                            this._processMessage(JSON.parse(nsPayload));
                            break;
                        // 68 = 'D' (a debug log).
                        case 68:
                            logger.debug(`[pid:${pid}] ${nsPayload.toString('utf8', 1)}`);
                            break;
                        // 87 = 'W' (a warn log).
                        case 87:
                            logger.warn(`[pid:${pid}] ${nsPayload.toString('utf8', 1)}`);
                            break;
                        // 69 = 'E' (an error log).
                        case 69:
                            logger.error(`[pid:${pid} ${nsPayload.toString('utf8', 1)}`);
                            break;
                        // 88 = 'X' (a dump log).
                        case 88:
                            // eslint-disable-next-line no-console
                            console.log(nsPayload.toString('utf8', 1));
                            break;
                        default:
                        ....省略
    
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
  _processMessage(msg) {
        if (msg.id) {
             ...省略
            if (msg.accepted) {
                logger.debug('request succeeded [method:%s, id:%s]', sent.method, sent.id);
                //确定消息处理完成，并在信令 Map表里确认并回调
                sent.resolve(msg.data);
            }
            else if (msg.error) {
                switch (msg.error) {
                    case 'TypeError':
                        sent.reject(new TypeError(msg.reason));
                        break;
                    default:
                        sent.reject(new Error(msg.reason));
                }
            }
            else {
                logger.error('received response is not accepted nor rejected [method:%s, id:%s]', sent.method, sent.id);
            }
        }
        // If a notification emit it to the corresponding entity.
        else if (msg.targetId && msg.event) {
            this.emit(msg.targetId, msg.event, msg.data);
        }
        // Otherwise unexpected message.
        else {
            logger.error('received message is not a response nor a notification');
        }
    }