node_flv_session.js

const URL = require("url");
const AMF = require("./node_core_amf");
const Logger = require("./node_core_logger");
const context = require("./node_core_ctx");
const NodeCoreUtils = require("./node_core_utils");

const FlvPacket = {
    create: (payload = null, type = 0, time = 0) => {
        return {
            header: {
                length: payload ? payload.length : 0,
                timestamp: time,
                type: type
            },
            payload: payload
        };
    }
};

class NodeFlvSession {
    constructor(config, req, res) {
        this.config = config;
        this.req = req;
        this.res = res;
        this.id = NodeCoreUtils.generateNewSessionID();
        this.ip = this.req.socket.remoteAddress;

        this.playStreamPath = "";
        this.playArgs = null;

        this.isStarting = false;
        this.isPlaying = false;
        this.isIdling = false;

        if (this.req.nmsConnectionType === "ws") {
            this.res.cork = this.res._socket.cork.bind(this.res._socket);
            this.res.uncork = this.res._socket.uncork.bind(this.res._socket);
            this.res.on("close", this.onReqClose.bind(this));
            this.res.on("error", this.onReqError.bind(this));
            this.res.write = this.res.send;
            this.res.end = this.res.close;
            this.TAG = "websocket-flv";
        } else {
            this.res.cork = this.res.socket.cork.bind(this.res.socket);
            this.res.uncork = this.res.socket.uncork.bind(this.res.socket);
            this.req.socket.on("close", this.onReqClose.bind(this));
            this.req.on("error", this.onReqError.bind(this));
            this.TAG = "http-flv";
        }

        this.numPlayCache = 0;
        context.sessions.set(this.id, this);
    }

    run() {
        let method = this.req.method;
        let urlInfo = URL.parse(this.req.url, true);
        let streamPath = urlInfo.pathname.split(".")[0];
        this.connectCmdObj = { ip: this.ip, method, streamPath, query: urlInfo.query };
        this.connectTime = new Date();
        this.isStarting = true;
        Logger.log(`[${this.TAG} connect] id=${this.id} ip=${this.ip} args=${JSON.stringify(urlInfo.query)}`);
        context.nodeEvent.emit("preConnect", this.id, this.connectCmdObj);
        if (!this.isStarting) {
            this.stop();
            return;
        }
        context.nodeEvent.emit("postConnect", this.id, this.connectCmdObj);

        if (method === "GET") {
            this.playStreamPath = streamPath;
            this.playArgs = urlInfo.query;

            this.onPlay();
        } else {
            this.stop();
        }
    }

    stop() {
        if (this.isStarting) {
            this.isStarting = false;
            let publisherId = context.publishers.get(this.playStreamPath);
            if (publisherId != null) {
                context.sessions.get(publisherId).players.delete(this.id);
                context.nodeEvent.emit("donePlay", this.id, this.playStreamPath, this.playArgs);
            }
            Logger.log(`[${this.TAG} play] Close stream. id=${this.id} streamPath=${this.playStreamPath}`);
            Logger.log(`[${this.TAG} disconnect] id=${this.id}`);
            context.nodeEvent.emit("doneConnect", this.id, this.connectCmdObj);
            this.res.end();
            context.idlePlayers.delete(this.id);
            context.sessions.delete(this.id);
        }
    }

    onReqClose() {
        this.stop();
    }

    onReqError(e) {
        this.stop();
    }

    reject() {
        Logger.log(`[${this.TAG} reject] id=${this.id}`);
        this.stop();
    }

    onPlay() {
        context.nodeEvent.emit("prePlay", this.id, this.playStreamPath, this.playArgs);
        if (!this.isStarting) {
            return;
        }
        if (this.config.auth !== undefined && this.config.auth.play) {
            let results = NodeCoreUtils.verifyAuth(this.playArgs.sign, this.playStreamPath, this.config.auth.secret);
            if (!results) {
                Logger.log(`[${this.TAG} play] Unauthorized. id=${this.id} streamPath=${this.playStreamPath} sign=${this.playArgs.sign}`);
                this.res.statusCode = 403;
                this.res.end();
                return;
            }
        }

        if (!context.publishers.has(this.playStreamPath)) {
            Logger.log(`[${this.TAG} play] Stream not found. id=${this.id} streamPath=${this.playStreamPath} `);
            context.idlePlayers.add(this.id);
            this.isIdling = true;
            return;
        }

        this.onStartPlay();
    }

    onStartPlay() {
        let publisherId = context.publishers.get(this.playStreamPath);
        let publisher = context.sessions.get(publisherId);
        let players = publisher.players;
        players.add(this.id);

        //send FLV header
        let FLVHeader = Buffer.from([0x46, 0x4c, 0x56, 0x01, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00]);
        if (publisher.isFirstAudioReceived) {
            FLVHeader[4] |= 0b00000100;
        }

        if (publisher.isFirstVideoReceived) {
            FLVHeader[4] |= 0b00000001;
        }
        this.res.write(FLVHeader);

        //send Metadata
        if (publisher.metaData != null) {
            let packet = FlvPacket.create(publisher.metaData, 18);
            let tag = NodeFlvSession.createFlvTag(packet);
            this.res.write(tag);
        }

        //send aacSequenceHeader
        if (publisher.audioCodec == 10) {
            let packet = FlvPacket.create(publisher.aacSequenceHeader, 8);
            let tag = NodeFlvSession.createFlvTag(packet);
            this.res.write(tag);
        }

        //send avcSequenceHeader
        if (publisher.videoCodec == 7 || publisher.videoCodec == 12) {
            let packet = FlvPacket.create(publisher.avcSequenceHeader, 9);
            let tag = NodeFlvSession.createFlvTag(packet);
            this.res.write(tag);
        }

        //send gop cache
        if (publisher.flvGopCacheQueue != null) {
            for (let tag of publisher.flvGopCacheQueue) {
                this.res.write(tag);
            }
        }

        this.isIdling = false;
        this.isPlaying = true;
        Logger.log(`[${this.TAG} play] Join stream. id=${this.id} streamPath=${this.playStreamPath} `);
        context.nodeEvent.emit("postPlay", this.id, this.playStreamPath, this.playArgs);
    }

    static createFlvTag(packet) {
        let PreviousTagSize = 11 + packet.header.length;
        let tagBuffer = Buffer.alloc(PreviousTagSize + 4);
        tagBuffer[0] = packet.header.type;
        tagBuffer.writeUIntBE(packet.header.length, 1, 3);
        tagBuffer[4] = (packet.header.timestamp >> 16) & 0xff;
        tagBuffer[5] = (packet.header.timestamp >> 8) & 0xff;
        tagBuffer[6] = packet.header.timestamp & 0xff;
        tagBuffer[7] = (packet.header.timestamp >> 24) & 0xff;
        tagBuffer.writeUIntBE(0, 8, 3);
        tagBuffer.writeUInt32BE(PreviousTagSize, PreviousTagSize);
        packet.payload.copy(tagBuffer, 11, 0, packet.header.length);
        return tagBuffer;
    }
}

module.exports = NodeFlvSession;