node-media-server: session.reject() not working

Question

I am trying to create an RTMP-server with the npm package: http://github.com/illuspas/Node-Media-Server. So the server works fine but I need to implement authentication in it. I am trying to check the authentication on "prePublish" event. I am querying the database and retrieving the user if the user was found then I want to let the user stream otherwise rejected. But the problem is, it doesn't leave it instead disconnects and then the stream automatically reconnected to it then it disconnects again and the loop goes on. How do I fix this problem?

Here is the code for the event:

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;

Here is the code of the entry point of the server:

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));

Here is the db file:

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;

My procedures to solve that problem:

1. Instead of the async function, I tried to handle the database query using a callback but it didn't work.
2. Before I was calling session.reject() now I am passing a callback there but the behavior is still the same

If you have any solution for that, please let me know. Thanks in advance

Answer 1

The actual prePublish call-back is not async, even if you assign an async function; while connecting to the database, the connection is accepted in the background, thus resulting in reconnect.

One way to fix this is to change the actual function that emits the prePublish, to async and write all your validation code there. Here is how:

1. you want to access your node modules and find node-media-server

2. under src/node_rtmp_session.js you will find a function called onPublish(invokeMessage) around line 1050. Change it to async.

3. after the emit event of "prePublish" add your validation code.

Finally, here is and example code of the updated function:

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);
    }
  }