feat: Measures end-to-end RTT with ping over the data channel. (#795)

* feat: Adds the user's region to presence. Fixes typos.

* feat: Measures end-to-end RTT with ping over the data channel.

JitsiConference.js

@@ -15,6 +15,7 @@ import {
 import AvgRTPStatsReporter from './modules/statistics/AvgRTPStatsReporter';
 import ComponentsVersions from './modules/version/ComponentsVersions';
 import ConnectionQuality from './modules/connectivity/ConnectionQuality';
+import E2ePing from './modules/e2eping/e2eping';
 import { getLogger } from 'jitsi-meet-logger';
 import GlobalOnErrorHandler from './modules/util/GlobalOnErrorHandler';
 import EventEmitter from 'events';
@@ -244,6 +245,33 @@ JitsiConference.prototype._init = function(options = {}) {
 
     this.room.updateDeviceAvailability(RTC.getDeviceAvailability());
 
+    this.e2eping = new E2ePing(
+        this.eventEmitter,
+        config,
+        (message, to) => {
+            try {
+                this.sendMessage(
+                    message, to, true /* sendThroughVideobridge */);
+            } catch (error) {
+                logger.warn('Failed to send a ping request or response.');
+            }
+        });
+    this.on(
+        JitsiConferenceEvents.USER_JOINED,
+        (id, participant) => this.e2eping.participantJoined(participant));
+    this.on(
+        JitsiConferenceEvents.USER_LEFT,
+        (id, participant) => this.e2eping.participantLeft(participant));
+    this.on(
+        JitsiConferenceEvents.ENDPOINT_MESSAGE_RECEIVED,
+        (participant, payload) => {
+            this.e2eping.messageReceived(participant, payload);
+        });
+    this.on(
+        JitsiConferenceEvents.DATA_CHANNEL_OPENED,
+        this.e2eping.dataChannelOpened);
+
+
     if (!this.rtc) {
         this.rtc = new RTC(this, options);
         this.eventManager.setupRTCListeners();
@@ -320,6 +348,11 @@ JitsiConference.prototype._init = function(options = {}) {
 
     // creates dominant speaker detection that works only in p2p mode
     this.p2pDominantSpeakerDetection = new P2PDominantSpeakerDetection(this);
+
+    if (config && config.deploymentInfo && config.deploymentInfo.userRegion) {
+        this.setLocalParticipantProperty(
+            'region', config.deploymentInfo.userRegion);
+    }
 };
 
 /**
@@ -621,7 +654,10 @@ JitsiConference.prototype.sendCommand = function(name, values) {
     if (this.room) {
         this.room.addToPresence(name, values);
         this.room.sendPresence();
+    } else {
+        logger.warn('Not sending a command, room not initialized.');
     }
+
 };
 
 /**

JitsiConferenceEvents.js

@@ -266,7 +266,7 @@ export const TRANSCRIPTION_STATUS_CHANGED
 
 
 /**
- * A new user joinned the conference.
+ * A new user joined the conference.
  */
 export const USER_JOINED = 'conference.userJoined';
 
@@ -286,6 +286,6 @@ export const USER_ROLE_CHANGED = 'conference.roleChanged';
 export const USER_STATUS_CHANGED = 'conference.statusChanged';
 
 /**
- * Event indicates that the bot participant type cahnged.
+ * Event indicates that the bot participant type changed.
  */
 export const BOT_TYPE_CHANGED = 'conference.bot_type_changed';

JitsiMeetJS.js

@@ -4,6 +4,7 @@ import { createGetUserMediaEvent } from './service/statistics/AnalyticsEvents';
 import AuthUtil from './modules/util/AuthUtil';
 import * as ConnectionQualityEvents
     from './service/connectivity/ConnectionQualityEvents';
+import * as E2ePingEvents from './service/e2eping/E2ePingEvents';
 import GlobalOnErrorHandler from './modules/util/GlobalOnErrorHandler';
 import * as JitsiConferenceErrors from './JitsiConferenceErrors';
 import * as JitsiConferenceEvents from './JitsiConferenceEvents';
@@ -139,7 +140,8 @@ export default _mergeNamespaceAndModule({
         connection: JitsiConnectionEvents,
         track: JitsiTrackEvents,
         mediaDevices: JitsiMediaDevicesEvents,
-        connectionQuality: ConnectionQualityEvents
+        connectionQuality: ConnectionQualityEvents,
+        e2eping: E2ePingEvents
     },
     errors: {
         conference: JitsiConferenceErrors,

modules/e2eping/e2eping.js

@@ -0,0 +1,334 @@
+/* global __filename */
+import { getLogger } from 'jitsi-meet-logger';
+import { createE2eRttEvent } from '../../service/statistics/AnalyticsEvents';
+import * as E2ePingEvents
+    from '../../service/e2eping/E2ePingEvents';
+import Statistics from '../statistics/statistics';
+
+const logger = getLogger(__filename);
+
+/**
+ * The 'type' of a message which designates an e2e ping request.
+ * @type {string}
+ */
+const E2E_PING_REQUEST = 'e2e-ping-request';
+
+/**
+ * The 'type' of a message which designates an e2e ping response.
+ * @type {string}
+ */
+const E2E_PING_RESPONSE = 'e2e-ping-response';
+
+/**
+ * Saves e2e ping related state for a single JitsiParticipant.
+ */
+class ParticipantWrapper {
+    /**
+     * Creates a ParticipantWrapper
+     * @param {JitsiParticipant} participant - The remote participant that this
+     * object wraps.
+     * @param {E2ePing} e2eping
+     */
+    constructor(participant, e2eping) {
+        // The JitsiParticipant
+        this.participant = participant;
+
+        // The E2ePing
+        this.e2eping = e2eping;
+
+        // Caches the ID
+        this.id = participant.getId();
+
+        // Recently sent requests
+        this.requests = {};
+
+        // The ID of the last sent request. We just increment it for each new
+        // request. Start at 1 so we can consider only thruthy values valid.
+        this.lastRequestId = 1;
+
+        this.clearIntervals = this.clearIntervals.bind(this);
+        this.sendRequest = this.sendRequest.bind(this);
+        this.handleResponse = this.handleResponse.bind(this);
+        this.maybeSendAnalytics = this.maybeSendAnalytics.bind(this);
+        this.sendAnalytics = this.sendAnalytics.bind(this);
+
+        // If the data channel was already open (this is likely a participant
+        // joining an existing conference) send a request immediately.
+        if (e2eping.isDataChannelOpen) {
+            this.sendRequest();
+        }
+
+        this.pingInterval = window.setInterval(
+            this.sendRequest, e2eping.pingIntervalMs);
+        this.analyticsInterval = window.setTimeout(
+            this.maybeSendAnalytics, this.e2eping.analyticsIntervalMs);
+    }
+
+    /**
+     * Clears the interval which sends pings.
+     * @type {*}
+     */
+    clearIntervals() {
+        if (this.pingInterval) {
+            window.clearInterval(this.pingInterval);
+        }
+        if (this.analyticsInterval) {
+            window.clearInterval(this.analyticsInterval);
+        }
+    }
+
+    /**
+     * Sends the next ping request.
+     * @type {*}
+     */
+    sendRequest() {
+        const requestId = this.lastRequestId++;
+        const requestMessage = {
+            type: E2E_PING_REQUEST,
+            id: requestId
+        };
+
+        this.e2eping.sendMessage(requestMessage, this.id);
+        this.requests[requestId] = {
+            id: requestId,
+            timeSent: window.performance.now()
+        };
+    }
+
+    /**
+     * Handles a response from this participant.
+     * @type {*}
+     */
+    handleResponse(response) {
+        const request = this.requests[response.id];
+
+        if (request) {
+            request.rtt = window.performance.now() - request.timeSent;
+            this.e2eping.eventEmitter.emit(
+                E2ePingEvents.E2E_RTT_CHANGED,
+                this.participant,
+                request.rtt);
+        }
+
+        this.maybeSendAnalytics();
+    }
+
+    /**
+     * Goes over the requests, clearing ones which we don't need anymore, and
+     * if it finds at least one request with a valid RTT in the last
+     * 'analyticsIntervalMs' then sends an analytics event.
+     * @type {*}
+     */
+    maybeSendAnalytics() {
+        const now = window.performance.now();
+
+        // The RTT we'll report is the minimum RTT measured in the last
+        // analyticsInterval
+        let rtt = Infinity;
+        let request, requestId;
+
+        // It's time to send analytics. Clean up all requests and find the
+        for (requestId in this.requests) {
+            if (this.requests.hasOwnProperty(requestId)) {
+                request = this.requests[requestId];
+
+                if (request.timeSent < now - this.e2eping.analyticsIntervalMs) {
+                    // An old request. We don't care about it anymore.
+                    delete this.requests[requestId];
+                } else if (request.rtt) {
+                    rtt = Math.min(rtt, request.rtt);
+                }
+            }
+        }
+
+        if (rtt < Infinity) {
+            this.sendAnalytics(rtt);
+        }
+    }
+
+    /**
+     * Sends an analytics event for this participant with the given RTT.
+     * @type {*}
+     */
+    sendAnalytics(rtt) {
+        Statistics.sendAnalytics(createE2eRttEvent(
+            this.id,
+            this.participant.getProperty('region'),
+            rtt));
+    }
+}
+
+/**
+ * Implements end-to-end ping (from one conference participant to another) via
+ * the jitsi-videobridge channel (either WebRTC data channel or web socket).
+ *
+ * TODO: use a broadcast message instead of individual pings to each remote
+ * participant.
+ *
+ * This class:
+ * 1. Sends periodic ping requests to all other participants in the
+ * conference.
+ * 2. Responds to ping requests from other participants.
+ * 3. Fires events with the end-to-end RTT to each participant whenever a
+ * response is received.
+ * 4. Fires analytics events with the end-to-end RTT periodically.
+ */
+export default class E2ePing {
+    /**
+     * @param {EventEmitter} eventEmitter - The object to use to emit events.
+     * @param {Function} sendMessage - The function to use to send a message.
+     * @param {Object} options
+     */
+    constructor(eventEmitter, options, sendMessage) {
+        this.eventEmitter = eventEmitter;
+        this.sendMessage = sendMessage;
+
+        // The interval at which pings will be sent (<= 0 disables sending).
+        this.pingIntervalMs = 10000;
+
+        // The interval at which analytics events will be sent.
+        this.analyticsIntervalMs = 60000;
+
+        // Maps a participant ID to its ParticipantWrapper
+        this.participants = {};
+
+        // Whether the WebRTC channel has been opened or not.
+        this.isDataChannelOpen = false;
+
+        if (options && options.e2eping) {
+            if (typeof options.e2eping.pingInterval === 'number') {
+                this.pingIntervalMs = options.e2eping.pingInterval;
+            }
+            if (typeof options.e2eping.analyticsInterval === 'number') {
+                this.analyticsIntervalMs = options.e2eping.analyticsInterval;
+            }
+
+            // We want to report at most once a ping interval.
+            if (this.analyticsIntervalMs > 0 && this.analyticsIntervalMs
+                < this.pingIntervalMs) {
+                this.analyticsIntervalMs = this.pingIntervalMs;
+            }
+        }
+        logger.info(
+            `Initializing e2e ping; pingInterval=${
+                this.pingIntervalMs}, analyticsInterval=${
+                this.analyticsIntervalMs}.`);
+    }
+
+    /**
+     * Notifies this instance that the communications channel has been opened
+     * and it can now send messages via sendMessage.
+     */
+    dataChannelOpened() {
+        this.isDataChannelOpen = true;
+
+        // We don't want to wait the whole interval before sending the first
+        // request, but we can't send it immediately after the participant joins
+        // either, because our data channel might not have initialized.
+        // So once the data channel initializes, send requests to everyone.
+        // Wait an additional 200ms to give a chance to the remote side (if it
+        // also just connected as is the case for the first 2 participants in a
+        // conference) to open its data channel.
+        for (const id in this.participants) {
+            if (this.participants.hasOwnProperty(id)) {
+                const participantWrapper = this.participants[id];
+
+                window.setTimeout(participantWrapper.sendRequest, 200);
+            }
+        }
+    }
+
+    /**
+     * Handles a message that was received.
+     *
+     * @param participant - The message sender.
+     * @param payload - The payload of the message.
+     */
+    messageReceived(participant, payload) {
+        // Listen to E2E PING requests and responses from other participants
+        // in the conference.
+        if (payload.type === E2E_PING_REQUEST) {
+            this.handleRequest(participant.getId(), payload);
+        } else if (payload.type === E2E_PING_RESPONSE) {
+            this.handleResponse(participant.getId(), payload);
+        }
+    }
+
+    /**
+     * Handles a participant joining the conference. Starts to send ping
+     * requests to the participant.
+     *
+     * @param {JitsiParticipant} participant - The participant that joined.
+     */
+    participantJoined(participant) {
+        const id = participant.getId();
+
+        if (this.pingIntervalMs <= 0) {
+            return;
+        }
+
+        if (this.participants[id]) {
+            logger.info(
+                `Participant wrapper already exists for ${id}. Clearing.`);
+            this.participants[id].clearIntervals();
+            delete this.participants[id];
+        }
+
+        this.participants[id] = new ParticipantWrapper(participant, this);
+    }
+
+    /**
+     * Handles a participant leaving the conference. Stops sending requests.
+     *
+     * @param {JitsiParticipant} participant - The participant that left.
+     */
+    participantLeft(participant) {
+        const id = participant.getId();
+
+        if (this.pingIntervalMs <= 0) {
+            return;
+        }
+
+        if (this.participants[id]) {
+            this.participants[id].clearIntervals();
+            delete this.participants[id];
+        }
+    }
+
+    /**
+     * Handles a ping request coming from another participant.
+     *
+     * @param {string} participantId - The ID of the participant who sent the
+     * request.
+     * @param {Object} request - The request.
+     */
+    handleRequest(participantId, request) {
+        // If it's a valid request, just send a response.
+        if (request && request.id) {
+            const response = {
+                type: E2E_PING_RESPONSE,
+                id: request.id
+            };
+
+            this.sendMessage(response, participantId);
+        } else {
+            logger.info(
+                `Received an invalid e2e ping request from ${participantId}.`);
+        }
+    }
+
+    /**
+     * Handles a ping response coming from another participant
+     * @param {string} participantId - The ID of the participant who sent the
+     * response.
+     * @param {Object} response - The response.
+     */
+    handleResponse(participantId, response) {
+        const participantWrapper = this.participants[participantId];
+
+        if (participantWrapper) {
+            participantWrapper.handleResponse(response);
+        }
+    }
+}
+

modules/xmpp/ChatRoom.js

@@ -570,7 +570,8 @@ export default class ChatRoom extends Listenable {
                 // seems there is some period of time in prosody that the
                 // configuration form is received but not applied. And if any
                 // participant joins during that period of time the first
-                // presence from the focus won't conain <item jid="focus..." />.
+                // presence from the focus won't contain
+                // <item jid="focus..." />.
                 memberOfThis.isFocus = true;
                 this._initFocus(from, jid);
             }

service/e2eping/E2ePingEvents.js

@@ -0,0 +1,4 @@
+/**
+ * Indicates that the end-to-end round-trip-time for a participant has changed.
+ */
+export const E2E_RTT_CHANGED = 'e2eping.e2e_rtt_changed';

service/statistics/AnalyticsEvents.js

@@ -17,7 +17,7 @@
  * constant. Otherwise use a factory function.
  *
  * Note that the AnalyticsAdapter uses the events passed to its functions for
- * its own purposes, and might modify them. Because of this factory functions
+ * its own purposes, and might modify them. Because of this, factory functions
  * should create new objects.
  *
  */
@@ -264,6 +264,27 @@ export const createConnectionStageReachedEvent = function(stage, attributes) {
     };
 };
 
+/**
+ * Creates an operational event for the end-to-end round trip time to a
+ * specific remote participant.
+ * @param participantId the ID of the remote participant.
+ * @param region the region of the remote participant
+ * @param rtt the rtt
+ */
+export const createE2eRttEvent = function(participantId, region, rtt) {
+    const attributes = {
+        'participant_id': participantId,
+        region,
+        rtt
+    };
+
+    return {
+        attributes,
+        name: 'e2e_rtt',
+        type: TYPE_OPERATIONAL
+    };
+};
+
 /**
  * Creates an event which indicates that the focus has left the MUC.
  */