zoneminder/web/js/EventStream.js

"use strict";

/**
 * EventStream - Manages a persistent zms MJPEG connection for event playback.
 *
 * Mirrors the MonitorStream.js constructor-function pattern. Frames arrive via
 * a hidden <img> receiving a multipart MJPEG stream from zms and are drawn to
 * a caller-supplied <canvas> on each img.onload.
 *
 * Commands (seek, pause, play, rate changes) are sent to zms over its existing
 * command-socket protocol via AJAX, exactly like MonitorStream and event.js.
 *
 * @param {Object} config
 * @param {number} config.monitorId
 * @param {number} config.monitorWidth   - Native monitor width
 * @param {number} config.monitorHeight  - Native monitor height
 * @param {string} config.url            - URL to index.php (for command AJAX)
 * @param {string} config.url_to_zms     - PathToZMS base URL
 * @param {HTMLCanvasElement} config.canvas
 * @param {number} [config.scale=100]    - Scale percentage
 */
function EventStream(config) {
  this.monitorId = config.monitorId;
  this.monitorWidth = config.monitorWidth;
  this.monitorHeight = config.monitorHeight;
  this.url = config.url;
  this.url_to_zms = config.url_to_zms;
  this.canvas = config.canvas;
  this.scale = config.scale ? parseInt(config.scale) : 100;

  this.connKey = null;
  this.img = null;
  this.started = false;
  this.paused = false;
  this.stopped = false;
  this.currentEventId = null;
  this.rate = 100;
  this.status = null;
  this.streamCmdTimer = null;
  this.ajaxQueue = null;
  this.rafId = null;
  this.bitmapPending = false;

  // Recovery state
  this.consecutiveErrors = 0;
  this.maxRecoveryAttempts = 5;
  this.recoveryDelay = 1000; // ms, doubles on each retry
  this.recoveryTimer = null;
  this.lastOptions = null; // saved for restart after recovery

  // Callbacks — set by the consumer
  this.onStatus = null;
  this.onError = null;

  // How often to poll zms for status (ms). Use the global if available,
  // otherwise fall back to a sensible default.
  this.statusInterval = (typeof statusRefreshTimeout !== 'undefined')
    ? statusRefreshTimeout
    : (typeof streamTimeout !== 'undefined') ? streamTimeout : 2000;

  // Command parameters template — matches MonitorStream / event.js protocol
  this.streamCmdParms = {
    view: 'request',
    request: 'stream',
    connkey: null
  };

  // -------------------------------------------------------------------------
  // connKey generation (identical to MonitorStream)
  // -------------------------------------------------------------------------

  this.genConnKey = function() {
    return (Math.floor((Math.random() * 999999) + 1))
        .toLocaleString('en-US', {minimumIntegerDigits: 6, useGrouping: false});
  };

  // -------------------------------------------------------------------------
  // start(eventId, options) — Begin streaming an event
  // -------------------------------------------------------------------------

  /**
   * @param {number|string} eventId
   * @param {Object}  [options]
   * @param {number}  [options.time]    - Epoch seconds to start at
   * @param {number}  [options.frame=1] - Frame ID to start at
   * @param {number}  [options.rate=100] - Playback rate (100 = 1x)
   * @param {string}  [options.replay='none']
   * @param {number}  [options.maxfps]  - Max FPS for the stream
   */
  this.start = function(eventId, options) {
    options = options || {};

    // An explicit start supersedes any pending recovery.
    if (this.recoveryTimer) {
      clearTimeout(this.recoveryTimer);
      this.recoveryTimer = null;
    }

    // Tear down any existing connection first. Re-using a live <img> would
    // abort its MJPEG stream behind zms's back and orphan the zms process.
    if (this.started || this.img) this.teardown(this.started);

    this.currentEventId = eventId;
    this.rate = (options.rate !== undefined) ? options.rate : 100;
    this.paused = false;
    this.stopped = false;
    this.lastOptions = Object.assign({}, options);

    // Fresh connkey for this stream
    this.connKey = this.genConnKey();
    this.streamCmdParms.connkey = this.connKey;

    // Build zms URL
    // UrlToZMS already carries '?monitor=N' for a monitor-specific stream,
    // so a second '?' here produced zms?monitor=24?source=event, which reads
    // as one giant monitor value and makes every stream log unreadable.
    var src = this.url_to_zms +
      (this.url_to_zms.indexOf('?') == -1 ? '?' : '&') +
      'source=event' +
      '&mode=jpeg' +
      '&event=' + eventId +
      '&monitor=' + this.monitorId +
      '&scale=' + this.scale +
      '&rate=' + this.rate +
      '&maxfps=' + (options.maxfps || 5) +
      '&replay=' + (options.replay || 'none') +
      '&connkey=' + this.connKey;

    if (options.frame) {
      src += '&frame=' + options.frame;
    }
    if (options.time) {
      src += '&time=' + options.time;
    }

    // Auth
    if (typeof zmAuth !== 'undefined') {
      src = zmAuth.appendTo(src);
    }

    // Use a DOM <img> element for MJPEG reception. Browsers natively
    // update a DOM <img> with each frame from a multipart/x-mixed-replace
    // response, but a detached Image() object does not reliably trigger
    // onload per frame.  We position it off-screen and draw from it to
    // the canvas on a requestAnimationFrame loop.
    if (!this.img) {
      this.img = document.createElement('img');
      this.img.style.cssText = 'position:absolute;left:-9999px;top:-9999px;' +
        'width:1px;height:1px;visibility:hidden;';
      document.body.appendChild(this.img);
    }

    var self = this;

    this.img.onerror = function() {
      console.warn('EventStream: MJPEG stream error for event ' +
        self.currentEventId + ' (monitor ' + self.monitorId + ')');
      self.streamCmdTimer = clearInterval(self.streamCmdTimer);
      if (self.rafId) {
        cancelAnimationFrame(self.rafId);
        self.rafId = null;
      }
      // Attempt recovery — zms likely died
      self.recover();
    };

    // Start the rAF draw loop — draws whenever the browser has
    // decoded a new MJPEG frame into the img element.
    this.startDrawLoop();

    // Setting src starts the MJPEG connection
    this.img.src = src;
    this.started = true;

    /* Poll for status from here, not from img.onload.
     *
     * onload is not a dependable per-restart signal for a
     * multipart/x-mixed-replace img: a restarted stream could paint its first
     * frame without ever getting a poller of its own, while the poller from
     * the connection before it kept running.  Starting the timer next to the
     * src that created the connkey ties the two together.
     *
     * The first query waits a full interval: zms creates its command socket
     * after the request reaches it, and asking before it exists returns
     * no_socket for a stream that is merely still starting.
     */
    this.streamCmdTimer = setInterval(
        this.streamCmdQuery.bind(this), this.statusInterval
    );
  };

  // -------------------------------------------------------------------------
  // teardown(quit) — Drop the connection, timers and draw loop.
  // Pass quit=false when zms is already gone, so we don't ask a dead process
  // to exit. Detaching the img handlers before clearing src keeps our own
  // teardown from firing onerror and looking like a stream failure.
  // -------------------------------------------------------------------------

  this.teardown = function(quit) {
    if (quit && this.started) this.streamCommand(CMD_QUIT);
    this.streamCmdTimer = clearInterval(this.streamCmdTimer);

    if (this.rafId) {
      cancelAnimationFrame(this.rafId);
      this.rafId = null;
    }

    if (this.img) {
      this.img.onload = null;
      this.img.onerror = null;
      this.img.src = '';
      if (this.img.parentNode) {
        this.img.parentNode.removeChild(this.img);
      }
      this.img = null;
    }
    this.bitmapPending = false;

    this.started = false;
    this.connKey = null;
    this.streamCmdParms.connkey = null;
  };

  // -------------------------------------------------------------------------
  // stop() — Stop the current stream
  // -------------------------------------------------------------------------

  this.stop = function() {
    if (this.recoveryTimer) {
      clearTimeout(this.recoveryTimer);
      this.recoveryTimer = null;
    }

    if (!this.started) return;

    this.teardown(true);
    this.paused = false;
    this.stopped = false;
    this.consecutiveErrors = 0;
    this.recoveryDelay = 1000;
  };

  // -------------------------------------------------------------------------
  // recover() — Attempt to restart after zms death
  // -------------------------------------------------------------------------

  this.recover = function() {
    // A recovery is already scheduled. Without this, every error arriving
    // while we wait to retry (the status poll keeps firing, and each reply
    // is another Error) would queue another restart and inflate the attempt
    // count until we give up on a stream that was never retried once.
    if (this.recoveryTimer) return;

    this.consecutiveErrors++;

    var self = this;
    var eventId = this.currentEventId;
    var opts = Object.assign({}, this.lastOptions || {});
    opts.rate = this.rate;

    // Drop the dead connection before deciding whether to retry, so that
    // giving up leaves nothing running. zms is already gone, so no CMD_QUIT.
    this.teardown(false);
    this.stopped = false;

    if (this.consecutiveErrors > this.maxRecoveryAttempts) {
      console.error('EventStream: max recovery attempts reached for monitor ' +
        this.monitorId + ', giving up');
      if (this.onError) this.onError('Stream recovery failed');
      return;
    }

    console.warn('EventStream: recovery attempt ' + this.consecutiveErrors +
      '/' + this.maxRecoveryAttempts + ' for monitor ' + this.monitorId);

    // Delay before restarting — exponential backoff
    this.recoveryTimer = setTimeout(function() {
      self.recoveryTimer = null;
      // teardown() cleared started, so a consumer polling for a live stream
      // may have restarted us while we waited. Restarting again here would
      // abort that healthy stream and orphan its zms, which fails the img
      // and lands us straight back in recover().
      if (self.started) return;
      self.start(eventId, opts);
    }, this.recoveryDelay);

    this.recoveryDelay = Math.min(this.recoveryDelay * 2, 10000);
  };

  // -------------------------------------------------------------------------
  // seek(offset) — Seek within the current event (seconds from start)
  // -------------------------------------------------------------------------

  this.seek = function(offset) {
    if (!this.started) return;
    this.streamCommand({command: CMD_SEEK, offset: offset});
  };

  // -------------------------------------------------------------------------
  // seekToTime(epochSecs) — Seek by wall-clock time
  // -------------------------------------------------------------------------

  this.seekToTime = function(epochSecs) {
    if (!this.started || !this.status) return;
    // status.event gives us the current event ID; we need the event's
    // start time to compute an offset.  If the caller hasn't provided
    // event metadata we fall back to duration-based estimation.
    //
    // For montagereview integration the caller will typically have the
    // event start time available in the global `events` object.
    var eventStartSecs = null;
    if (typeof events !== 'undefined' && events[this.currentEventId]) {
      eventStartSecs = events[this.currentEventId].StartTimeSecs;
    }
    if (eventStartSecs) {
      var offset = epochSecs - eventStartSecs;
      if (offset < 0) offset = 0;
      this.seek(offset);
    }
  };

  // -------------------------------------------------------------------------
  // setRate(rate) — Change playback rate (100 = 1x realtime)
  // -------------------------------------------------------------------------

  this.setRate = function(rate) {
    this.rate = rate;
    if (!this.started) return;
    this.streamCommand({command: CMD_VARPLAY, rate: rate});
  };

  // -------------------------------------------------------------------------
  // pause() / play()
  // -------------------------------------------------------------------------

  this.pause = function() {
    if (!this.started) return;
    this.paused = true;
    this.streamCommand(CMD_PAUSE);
  };

  this.play = function() {
    if (!this.started) return;
    this.paused = false;
    this.streamCommand(CMD_PLAY);
  };

  // -------------------------------------------------------------------------
  // setScale(scale) — Change the stream scale
  // -------------------------------------------------------------------------

  this.setScale = function(scale) {
    this.scale = scale;
    if (!this.started) return;
    this.streamCommand({command: CMD_SCALE, scale: scale});
  };

  // -------------------------------------------------------------------------
  // switchEvent(eventId, options) — Switch to a different event
  // -------------------------------------------------------------------------

  this.switchEvent = function(eventId, options) {
    if (this.recoveryTimer) {
      clearTimeout(this.recoveryTimer);
      this.recoveryTimer = null;
    }

    // Reset recovery state for fresh event
    this.consecutiveErrors = 0;
    this.recoveryDelay = 1000;

    // start() tears the old stream down (sending CMD_QUIT) and brings the new
    // event up in one step. Doing the teardown here instead and starting from
    // a timer would leave started=false in between, and a consumer polling for
    // a live stream would start its own before the timer fired.
    this.start(eventId, options);
  };

  // -------------------------------------------------------------------------
  // streamCommand(command) — Send a command to zms via AJAX
  // -------------------------------------------------------------------------

  this.streamCommand = function(command) {
    if (!this.started) {
      return;
    }
    var params = Object.assign({}, this.streamCmdParms);
    if (typeof command === 'object') {
      for (var key in command) {
        if (command.hasOwnProperty(key)) params[key] = command[key];
      }
    } else {
      params.command = command;
    }
    this.streamCmdReq(params);
  };

  // -------------------------------------------------------------------------
  // streamCmdReq(params) — Low-level AJAX to the command socket
  // -------------------------------------------------------------------------

  this.streamCmdReq = function(params) {
    var self = this;
    // The connkey this exchange is about. Replies are queued and can land
    // after a restart has replaced it; see getStreamCmdResponse().
    var connKey = params.connkey;
    this.ajaxQueue = jQuery.ajaxQueue({
      url: zmAuth.appendTo(this.url),
      xhrFields: {withCredentials: true},
      data: params,
      dataType: 'json'
    })
        .done(function(respObj) {
          self.getStreamCmdResponse(respObj, connKey);
        })
        .fail(function(jqXHR, textStatus) {
          if (textStatus === 'abort') return;
          if (connKey !== self.connKey) return;
          console.warn('EventStream: AJAX failed for monitor ' +
            self.monitorId + ': ' + textStatus);
          // AJAX failure likely means zms has died (socket gone)
          self.recover();
        });
  };

  // -------------------------------------------------------------------------
  // streamCmdQuery() — Periodic CMD_QUERY for status updates
  // -------------------------------------------------------------------------

  this.streamCmdQuery = function() {
    if (this.started) {
      var params = Object.assign({}, this.streamCmdParms);
      params.command = CMD_QUERY;
      this.streamCmdReq(params);
    }
  };

  // -------------------------------------------------------------------------
  // getStreamCmdResponse(respObj) — Handle CMD_QUERY / command responses
  // -------------------------------------------------------------------------

  this.getStreamCmdResponse = function(respObj, connKey) {
    if (!respObj) return;

    /* A reply for a connkey we no longer hold describes a stream that is
     * already gone: it was issued before a restart and queued behind other
     * requests.  Acting on it restarted the healthy stream that replaced it,
     * which produced another stale reply, and so on - the stream never lived
     * long enough to deliver a second frame.
     */
    if (connKey !== undefined && connKey !== this.connKey) return;

    if (respObj.result === 'Error' || respObj.result === 'Err') {
      console.warn('EventStream: command error for monitor ' +
        this.monitorId + ': ' + respObj.message);
      if (!EventStream.errorIsFatal(respObj.reason)) {
        // zms is alive, this one exchange failed. Retry on the next poll.
        return;
      }
      this.recover();
      return;
    }

    // Successful response — reset error counter
    this.consecutiveErrors = 0;
    this.recoveryDelay = 1000;

    if (!respObj.status) return;

    this.status = respObj.status;

    // Update the credential if the server sent a fresh one
    if (typeof zmAuth !== 'undefined') {
      zmAuth.update(this.status);
    }

    // Track paused and stopped state from server
    if (this.status.paused !== undefined) {
      this.paused = !!this.status.paused;
    }
    if (this.status.stopped !== undefined) {
      this.stopped = !!this.status.stopped;
    }

    // Notify consumer
    if (this.onStatus) {
      this.onStatus(this.status);
    }
  };

  // -------------------------------------------------------------------------
  // startDrawLoop() — rAF loop that copies the MJPEG img to the canvas
  // -------------------------------------------------------------------------

  this.startDrawLoop = function() {
    var self = this;
    function loop() {
      if (!self.started) return;
      self.drawFrame();
      self.rafId = requestAnimationFrame(loop);
    }
    this.rafId = requestAnimationFrame(loop);
  };

  // -------------------------------------------------------------------------
  // drawFrame() — Draw the current MJPEG frame to the canvas
  // -------------------------------------------------------------------------

  this.drawFrame = function() {
    if (!this.canvas || !this.img) return;
    // Only draw if the img has decoded at least one frame
    if (!this.img.naturalWidth) return;

    // Gecko gives canvas drawImage() the frame the img held when it was first
    // decoded and never refreshes it, so copying a multipart/x-mixed-replace
    // img straight to the canvas repaints that one frame forever while zms
    // goes on streaming: montage review froze on its first frame in Firefox
    // while playing normally in Chrome.  createImageBitmap() reads the frame
    // the element is showing now.  It is async, so skip a tick while one is
    // outstanding instead of queueing requests up behind the stream rate.
    if (this.bitmapPending) return;
    var self = this;
    this.bitmapPending = true;
    createImageBitmap(this.img).then(function(bitmap) {
      self.bitmapPending = false;
      if (!self.canvas) {
        bitmap.close();
        return;
      }
      var ctx = self.canvas.getContext('2d');
      ctx.drawImage(bitmap, 0, 0, self.canvas.width, self.canvas.height);
      bitmap.close();
      if (self.onFrameDrawn) self.onFrameDrawn(self.canvas);
    }).catch(function() {
      // A frame that arrives mid-teardown, or one that is only partly
      // received, is not worth reporting: the next tick draws the next frame.
      self.bitmapPending = false;
    });
  };
}

/* Does this ajax/stream.php failure mean the zms behind our connkey is gone?
 *
 * Only then is restarting right: a restart replaces the connkey and leaves any
 * still-running zms unaddressable. A slow reply or a php-local socket problem
 * says nothing about zms.  An absent reason means a php that predates the
 * field, so keep the older always-restart behaviour.
 *
 * Same rule as streamErrorIsFatal() in MonitorStream.js, kept here because the
 * views that use EventStream do not load MonitorStream.js.
 */
EventStream.errorIsFatal = function(reason) {
  if (!reason) return true;
  return reason == 'no_socket';
};

if (typeof module !== 'undefined' && module.exports) {
  module.exports = {EventStream};
}