节点 - 媒体服务器：session.reject（）不起作用

Question

我正在尝试使用NPM软件包创建一个RTMP-Server： http://github.com/ Illuspas/Node-Media-Server 。因此，服务器工作正常，但我需要在其中实现身份验证。我正在尝试检查“ prefublish”事件的身份验证。我正在查询数据库并检索用户，如果找到了用户，我想让用户流其他拒绝。但是问题是，它不会离开它而是断开连接，然后流自动重新连接到它，然后再次断开连接，循环继续进行。如何解决此问题？

这是事件的代码：

const NodeMediaServer = require('node-media-server');
const config = require('./config').rtmp_server;
const db = require('./db');

const nms = new NodeMediaServer(config);

const getStreamKeyFromStreamPath = (path) => {
  const parts = path.split('/');
  return parts[parts.length - 1];
};

nms.on('prePublish', async (id, StreamPath, args) => {
  const session = nms.getSession(id);
  try {
    const streamKey = getStreamKeyFromStreamPath(StreamPath);

    const validStream = (
      await db.query('SELECT * FROM public."People" WHERE stream_key = $1', [streamKey])
    ).rows[0];

    console.log(validStream);

    if (validStream) {
      // do stuff
    } else {
      session.reject((reason) => {
        console.log(reason);
      });
    }

    console.log(
      '[NodeEvent on prePublish]',
      `id=${id} StreamPath=${StreamPath} args=${JSON.stringify(args)}`
    );
  } catch (err) {
    session.reject();
  }
});

module.exports = nms;

以下是服务器入口点的代码：

require("dotenv").config();
const db = require("./db");
const nms = require("./nms");

// database connection
db.connect()
  .then(() => {
    console.log("Connected to database");
    // start the rtmp server
    nms.run();
  })
  .catch((err) => console.log(err.message));

这是DB文件：

const { Pool } = require('pg');

const connectionString = process.env.PG_CONNECTION_STRING;

const poolOptions = {
  host: process.env.PG_HOST,
  user: process.env.PG_USER,
  port: process.env.PG_PORT,
  password: process.env.PG_PASSWORD,
  database: process.env.PG_DATABASE,
};

const pool = new Pool(process.env.NODE_ENV === 'production' ? connectionString : poolOptions);

module.exports = pool;

我解决该问题的过程：

1. 我试图使用回调来处理数据库查询，而不是异步函数，但是它不起作用。
2. 在调用session.reject（）现在，我正在那里通过一个回调，但是

如果您对此有任何解决方案，则行为仍然相同，请告诉我。 提前致谢

Answer 1

即使您分配异步函数，实际的profublish呼叫背包也不是异步。连接到数据库时，该连接在后台接受，从而导致重新连接。

解决此问题的一种方法是更改​​排放前出版物的实际功能，异步并在此处编写所有验证代码。以下是：

1. 您想访问节点模块并找到节点 - 媒体服务器

2. 在src/node_rtmp_session.js下找到节点 - 媒体 - server

   您会找到一个称为inpublish（InvokeMessage）围绕第1050行的函数。 /p>

3. 在发射“ prefublish”的发射事件后，添加了您的验证代码。

最后，这是更新函数的示例代码：

async onPublish(invokeMessage) {
    if (typeof invokeMessage.streamName !== 'string') {
      return;
    }
    this.publishStreamPath = '/' + this.appname + '/' + invokeMessage.streamName.split('?')[0];
    this.publishArgs = QueryString.parse(invokeMessage.streamName.split('?')[1]);
    this.publishStreamId = this.parserPacket.header.stream_id;
    context.nodeEvent.emit('prePublish', this.id, this.publishStreamPath, this.publishArgs);

    //  YOUR VALIDATION CODE HERE
    const streamKey = getStreamKeyFromStreamPath(StreamPath);

    const validStream = (
      await db.query('SELECT * FROM public."People" WHERE stream_key = $1', [streamKey])
    ).rows[0];

    console.log(validStream);

    if (validStream) {
      // do stuff
    } else {
      this.reject();
      console.log("session rejected");
    }

    // YOUR VALIDATION CODE HERE

    if (!this.isStarting) {
      return;
    }

    if (this.config.auth && this.config.auth.publish && !this.isLocal) {
      let results = NodeCoreUtils.verifyAuth(this.publishArgs.sign, this.publishStreamPath, this.config.auth.secret);
      if (!results) {
        Logger.log(`[rtmp publish] Unauthorized. id=${this.id} streamPath=${this.publishStreamPath} streamId=${this.publishStreamId} sign=${this.publishArgs.sign} `);
        this.sendStatusMessage(this.publishStreamId, 'error', 'NetStream.publish.Unauthorized', 'Authorization required.');
        return;
      }
    }

    if (context.publishers.has(this.publishStreamPath)) {
      this.reject();
      Logger.log(`[rtmp publish] Already has a stream. id=${this.id} streamPath=${this.publishStreamPath} streamId=${this.publishStreamId}`);
      this.sendStatusMessage(this.publishStreamId, 'error', 'NetStream.Publish.BadName', 'Stream already publishing');
    } else if (this.isPublishing) {
      Logger.log(`[rtmp publish] NetConnection is publishing. id=${this.id} streamPath=${this.publishStreamPath} streamId=${this.publishStreamId}`);
      this.sendStatusMessage(this.publishStreamId, 'error', 'NetStream.Publish.BadConnection', 'Connection already publishing');
    } else {
      Logger.log(`[rtmp publish] New stream. id=${this.id} streamPath=${this.publishStreamPath} streamId=${this.publishStreamId}`);
      context.publishers.set(this.publishStreamPath, this.id);
      this.isPublishing = true;

      this.sendStatusMessage(this.publishStreamId, 'status', 'NetStream.Publish.Start', `${this.publishStreamPath} is now published.`);
      for (let idlePlayerId of context.idlePlayers) {
        let idlePlayer = context.sessions.get(idlePlayerId);
        if (idlePlayer && idlePlayer.playStreamPath === this.publishStreamPath) {
          idlePlayer.onStartPlay();
          context.idlePlayers.delete(idlePlayerId);
        }
      }
      context.nodeEvent.emit('postPublish', this.id, this.publishStreamPath, this.publishArgs);
    }
  }