зеркало из https://github.com/mozilla/gecko-dev.git
Bug 1028869 - Part 1: Add ping and ack operations to PushHandler. r=standard8
This commit is contained in:
Paul Kerr [:pkerr]
Родитель
6dc4bab370
Коммит
9acb1ef084
|
|
@ -1660,6 +1660,8 @@ pref("loop.do_not_disturb", false);
|
|||
pref("loop.ringtone", "chrome://browser/content/loop/shared/sounds/ringtone.ogg");
|
||||
pref("loop.retry_delay.start", 60000);
|
||||
pref("loop.retry_delay.limit", 300000);
|
||||
pref("loop.ping.interval", 1800000);
|
||||
pref("loop.ping.timeout", 10000);
|
||||
pref("loop.feedback.baseUrl", "https://input.mozilla.org/api/v1/feedback");
|
||||
pref("loop.feedback.product", "Loop");
|
||||
pref("loop.debug.loglevel", "Error");
|
||||
|
|
|
|||
|
|
@ -9,12 +9,308 @@ const { classes: Cc, interfaces: Ci, utils: Cu } = Components;
|
|||
Cu.import("resource://gre/modules/Services.jsm");
|
||||
Cu.import("resource://gre/modules/XPCOMUtils.jsm");
|
||||
Cu.import("resource://gre/modules/Timer.jsm");
|
||||
Cu.import("resource://gre/modules/Promise.jsm");
|
||||
|
||||
const {MozLoopService} = Cu.import("resource:///modules/loop/MozLoopService.jsm", {});
|
||||
const consoleLog = MozLoopService.log;
|
||||
|
||||
this.EXPORTED_SYMBOLS = ["MozLoopPushHandler"];
|
||||
|
||||
XPCOMUtils.defineLazyModuleGetter(this, "console",
|
||||
"resource://gre/modules/devtools/Console.jsm");
|
||||
const CONNECTION_STATE_CLOSED = 0;
|
||||
const CONNECTION_STATE_CONNECTING = 1;
|
||||
const CONNECTION_STATE_OPEN = 2;
|
||||
|
||||
const SERVICE_STATE_OFFLINE = 0;
|
||||
const SERVICE_STATE_PENDING = 1;
|
||||
const SERVICE_STATE_ACTIVE = 2;
|
||||
|
||||
function PushSocket(webSocket = null) {
|
||||
this._websocket = webSocket;
|
||||
}
|
||||
|
||||
PushSocket.prototype = {
|
||||
|
||||
/**
|
||||
* Open push-notification websocket.
|
||||
*
|
||||
* @param {String} pushUri
|
||||
* @param {Function} onMsg(aMsg) callback receives any incoming messages
|
||||
* aMsg is constructed from the json payload; both
|
||||
* text and binary message reception are mapped to this
|
||||
* callback.
|
||||
* @param {Function} onStart called when the socket is connected
|
||||
* @param {Function} onClose(aCode, aReason) called when the socket closes;
|
||||
* both near and far side close events map to this
|
||||
* callback.
|
||||
* aCode is any status code returned on close
|
||||
* aReason is any string returned on close
|
||||
*/
|
||||
|
||||
connect: function(pushUri, onMsg, onStart, onClose) {
|
||||
if (!pushUri || !onMsg || !onStart || !onClose) {
|
||||
throw new Error("PushSocket: missing required parameter(s):"
|
||||
(pushUri ? "" : " pushUri") +
|
||||
(onMsg ? "" : " onMsg") +
|
||||
(onStart ? "" : " onStart") +
|
||||
(onClose ? "" : " onClose"));
|
||||
}
|
||||
|
||||
this._onMsg = onMsg;
|
||||
this._onStart = onStart;
|
||||
this._onClose = onClose;
|
||||
|
||||
if (!this._websocket) {
|
||||
this._websocket = Cc["@mozilla.org/network/protocol;1?name=wss"]
|
||||
.createInstance(Ci.nsIWebSocketChannel);
|
||||
}
|
||||
|
||||
let uri = Services.io.newURI(pushUri, null, null);
|
||||
this._websocket.protocol = "push-notification";
|
||||
this._websocket.asyncOpen(uri, pushUri, this, null);
|
||||
},
|
||||
|
||||
/**
|
||||
* nsIWebSocketListener method, handles the start of the websocket stream.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
*/
|
||||
onStart: function() {
|
||||
this._socketOpen = true;
|
||||
this._onStart();
|
||||
},
|
||||
|
||||
/**
|
||||
* nsIWebSocketListener method, called when the websocket is closed locally.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {nsresult} aStatusCode
|
||||
*/
|
||||
onStop: function(aContext, aStatusCode) {
|
||||
this._socketOpen = false;
|
||||
this._onClose(aStatusCode, "websocket onStop");
|
||||
},
|
||||
|
||||
/**
|
||||
* nsIWebSocketListener method, called when the websocket is closed
|
||||
* by the far end.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {integer} aCode the websocket closing handshake close code
|
||||
* @param {String} aReason the websocket closing handshake close reason
|
||||
*/
|
||||
onServerClose: function(aContext, aCode, aReason) {
|
||||
this._socketOpen = false;
|
||||
this._onClose(aCode, aReason);
|
||||
},
|
||||
|
||||
/**
|
||||
* nsIWebSocketListener method, called when the websocket receives
|
||||
* a text message (normally json encoded).
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {String} aMsg The message data
|
||||
*/
|
||||
onMessageAvailable: function(aContext, aMsg) {
|
||||
consoleLog.log("PushSocket: Message received: ", aMsg);
|
||||
if (!this._socketOpen) {
|
||||
consoleLog.error("Message received in Winsocket closed state");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this._onMsg(JSON.parse(aMsg));
|
||||
}
|
||||
catch (error) {
|
||||
consoleLog.error("PushSocket: error parsing message payload - ", error);
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* nsIWebSocketListener method, called when the websocket receives a binary message.
|
||||
* This class assumes that it is connected to a SimplePushServer and therefore treats
|
||||
* the message payload as json encoded.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {String} aMsg The message data
|
||||
*/
|
||||
onBinaryMessageAvailable: function(aContext, aMsg) {
|
||||
consoleLog.log("PushSocket: Binary message received: ", aMsg);
|
||||
if (!this._socketOpen) {
|
||||
consoleLog.error("PushSocket: message receive in Winsocket closed state");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this._onMsg(JSON.parse(aMsg));
|
||||
}
|
||||
catch (error) {
|
||||
consoleLog.error("PushSocket: error parsing message payload - ", error);
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Create a JSON encoded message payload and send via websocket.
|
||||
*
|
||||
* @param {Object} aMsg Message to send.
|
||||
*
|
||||
* @returns {Boolean} true if message has been sent, false otherwise
|
||||
*/
|
||||
send: function(aMsg) {
|
||||
if (!this._socketOpen) {
|
||||
consoleLog.error("PushSocket: attempt to send before websocket is open");
|
||||
return false;
|
||||
}
|
||||
|
||||
let msg;
|
||||
try {
|
||||
msg = JSON.stringify(aMsg);
|
||||
}
|
||||
catch (error) {
|
||||
consoleLog.error("PushSocket: JSON generation error - ", error);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
this._websocket.sendMsg(msg);
|
||||
consoleLog.log("PushSocket: Message sent: ", msg);
|
||||
}
|
||||
// guard against the case that the websocket has closed before this call.
|
||||
catch (e) {
|
||||
consoleLog.warn("PushSocket: websocket send error", e);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
},
|
||||
|
||||
/**
|
||||
* Close the websocket.
|
||||
*/
|
||||
close: function() {
|
||||
if (!this._socketOpen) {
|
||||
return;
|
||||
}
|
||||
|
||||
this._socketOpen = false;
|
||||
consoleLog.info("PushSocket: websocket closing");
|
||||
|
||||
// Do not pass through any callbacks after this point.
|
||||
this._onStart = function() {};
|
||||
this._onMsg = this._onStart;
|
||||
this._onClose = this._onStart;
|
||||
|
||||
try {
|
||||
this._websocket.close(this._websocket.CLOSE_NORMAL);
|
||||
}
|
||||
catch (e) {}
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
/**
|
||||
* Create a RetryManager object. Class to handle retrying a UserAgent
|
||||
* to PushServer request following a retry back-off scheme managed by
|
||||
* this class. The current delay mechanism is to double the delay
|
||||
* each time an operation to be retried until a maximum is met.
|
||||
*
|
||||
* @param {Integer} startDelay The initial delay interval in milliseconds.
|
||||
* @param {Integer} maxDelay Maximum time delay value in milliseconds.
|
||||
*/
|
||||
function RetryManager (startDelay, maxDelay) {
|
||||
if (!startDelay || !maxDelay) {
|
||||
throw new Error("RetryManager: missing required parameters(s)" +
|
||||
(startDelay ? "" : " startDelay") +
|
||||
(maxDelay ? "" : " maxDelay"));
|
||||
}
|
||||
|
||||
this._startDelay = startDelay;
|
||||
// The maximum delay cannot be less than the starting delay.
|
||||
this._maxDelay = maxDelay > startDelay ? maxDelay : startDelay;
|
||||
}
|
||||
|
||||
RetryManager.prototype = {
|
||||
/**
|
||||
* Method to handle retrying a UserAgent to PushServer request.
|
||||
*
|
||||
* @param {Function} delayedOp Function to call after current delay is satisfied
|
||||
*/
|
||||
retry: function(delayedOp) {
|
||||
if (!this._timeoutID) {
|
||||
this._retryDelay = this._startDelay;
|
||||
} else {
|
||||
clearTimeout(this._timeoutID);
|
||||
let nextDelay = this._retryDelay * 2;
|
||||
this._retryDelay = nextDelay > this._maxDelay ? this._maxDelay : nextDelay;
|
||||
}
|
||||
|
||||
this._timeoutID = setTimeout(delayedOp, this._retryDelay);
|
||||
consoleLog.log("PushHandler: retry delay set for ", this._retryDelay);
|
||||
},
|
||||
|
||||
/**
|
||||
* Method used to reset the delay back-off logic and clear any currently
|
||||
* running delay timeout.
|
||||
*/
|
||||
reset: function() {
|
||||
if (this._timeoutID) {
|
||||
clearTimeout(this._timeoutID);
|
||||
this._timeoutID = null;
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
* Create a PingMonitor object. An object instance will periodically execute
|
||||
* a ping send function and if not reset, will then execute an error function.
|
||||
*
|
||||
* @param {Function} pingFunc Function that is called after a ping interval
|
||||
* has expired without being restart.
|
||||
* @param {Function} onTimeout Function that is called after a ping timeout
|
||||
* interval has expired without restart being called.
|
||||
* @param {Integer} interval Timeout value in milliseconds between successive
|
||||
* pings or between the last restart call and a ping.
|
||||
* When this interval expires, pingFunc is called and the
|
||||
* timeout interval is started.
|
||||
* @param {Integer} timeout Timeout value in milliseconds between a call to
|
||||
* pingFunc and a call to onTimeout unless restart is called.
|
||||
* Restart will begin the ping timeout interval again.
|
||||
*/
|
||||
function PingMonitor(pingFunc, onTimeout, interval, timeout) {
|
||||
if (!pingFunc || !onTimeout || !interval || !timeout) {
|
||||
throw new Error("PingMonitor: missing required parameters");
|
||||
}
|
||||
this._onTimeout = onTimeout;
|
||||
this._pingFunc = pingFunc;
|
||||
this._pingInterval = interval;
|
||||
this._pingTimeout = timeout;
|
||||
}
|
||||
|
||||
PingMonitor.prototype = {
|
||||
/**
|
||||
* Function to restart the ping timeout and cancel any current timeout operation.
|
||||
*/
|
||||
restart: function () {
|
||||
consoleLog.info("PushHandler: ping timeout restart");
|
||||
this.stop();
|
||||
this._pingTimerID = setTimeout(() => {this._pingSend()}, this._pingInterval);
|
||||
},
|
||||
|
||||
/**
|
||||
* Function to stop the PingMonitor.
|
||||
*/
|
||||
stop: function() {
|
||||
if (this._pingTimerID){
|
||||
clearTimeout(this._pingTimerID);
|
||||
this._pingTimerID = undefined;
|
||||
}
|
||||
},
|
||||
|
||||
_pingSend: function () {
|
||||
consoleLog.info("PushHandler: ping sent");
|
||||
this._pingTimerID = setTimeout(this._onTimeout, this._pingTimeout);
|
||||
this._pingFunc();
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
/**
|
||||
* We don't have push notifications on desktop currently, so this is a
|
||||
|
|
@ -25,31 +321,53 @@ let MozLoopPushHandler = {
|
|||
pushServerUri: undefined,
|
||||
// Records containing the registration and notification callbacks indexed by channelID.
|
||||
// Each channel will be registered with the PushServer.
|
||||
channels: {},
|
||||
channels: new Map(),
|
||||
// This is the UserAgent UUID assigned by the PushServer
|
||||
uaID: undefined,
|
||||
// Each successfully registered channelID is used as a key to hold its pushEndpoint URL.
|
||||
registeredChannels: {},
|
||||
// Push protocol state variable
|
||||
serviceState: SERVICE_STATE_OFFLINE,
|
||||
// Websocket connection state variable
|
||||
connectionState: CONNECTION_STATE_CLOSED,
|
||||
// Contains channels that need to be registered with the PushServer
|
||||
_channelsToRegister: [],
|
||||
|
||||
_channelsToRegister: {},
|
||||
|
||||
_minRetryDelay_ms: (() => {
|
||||
get _startRetryDelay_ms() {
|
||||
try {
|
||||
return Services.prefs.getIntPref("loop.retry_delay.start")
|
||||
return Services.prefs.getIntPref("loop.retry_delay.start");
|
||||
}
|
||||
catch (e) {
|
||||
return 60000 // 1 minute
|
||||
return 60000; // 1 minute
|
||||
}
|
||||
})(),
|
||||
},
|
||||
|
||||
_maxRetryDelay_ms: (() => {
|
||||
get _maxRetryDelay_ms() {
|
||||
try {
|
||||
return Services.prefs.getIntPref("loop.retry_delay.limit")
|
||||
return Services.prefs.getIntPref("loop.retry_delay.limit");
|
||||
}
|
||||
catch (e) {
|
||||
return 300000 // 5 minutes
|
||||
return 300000; // 5 minutes
|
||||
}
|
||||
})(),
|
||||
},
|
||||
|
||||
get _pingInterval_ms() {
|
||||
try {
|
||||
return Services.prefs.getIntPref("loop.ping.interval");
|
||||
}
|
||||
catch (e) {
|
||||
return 18000000; // 30 minutes
|
||||
}
|
||||
},
|
||||
|
||||
get _pingTimeout_ms() {
|
||||
try {
|
||||
return Services.prefs.getIntPref("loop.ping.timeout");
|
||||
}
|
||||
catch (e) {
|
||||
return 10000; // 10 seconds
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Inializes the PushHandler and opens a socket with the PushServer.
|
||||
|
|
@ -61,8 +379,9 @@ let MozLoopPushHandler = {
|
|||
* used for testing.
|
||||
*/
|
||||
initialize: function(options = {}) {
|
||||
consoleLog.info("PushHandler: initialize options = ", options);
|
||||
if (Services.io.offline) {
|
||||
console.warn("MozLoopPushHandler - IO offline");
|
||||
consoleLog.warn("PushHandler: IO offline");
|
||||
return false;
|
||||
}
|
||||
|
||||
|
|
@ -71,6 +390,14 @@ let MozLoopPushHandler = {
|
|||
}
|
||||
|
||||
this._initDone = true;
|
||||
this._retryManager = new RetryManager(this._startRetryDelay_ms,
|
||||
this._maxRetryDelay_ms);
|
||||
// Send an empty json payload as a ping.
|
||||
// Close the websocket and re-open if a timeout occurs.
|
||||
this._pingMonitor = new PingMonitor(() => this._pushSocket.send({}),
|
||||
() => this._restartConnection(),
|
||||
this._pingInterval_ms,
|
||||
this._pingTimeout_ms);
|
||||
|
||||
if ("mockWebSocket" in options) {
|
||||
this._mockWebSocket = options.mockWebSocket;
|
||||
|
|
@ -80,6 +407,42 @@ let MozLoopPushHandler = {
|
|||
return true;
|
||||
},
|
||||
|
||||
/**
|
||||
* Reset and clear PushServer connection.
|
||||
* Returns MozLoopPushHandler to pre-initialized state.
|
||||
*/
|
||||
shutdown: function() {
|
||||
consoleLog.info("PushHandler: shutdown");
|
||||
if (!this._initDone) {
|
||||
return;
|
||||
}
|
||||
|
||||
this._initDone = false;
|
||||
this._retryManager.reset();
|
||||
this._pingMonitor.stop();
|
||||
|
||||
// Un-register each active notification channel
|
||||
if (this.connectionState === CONNECTION_STATE_OPEN) {
|
||||
Object.keys(this.registeredChannels).forEach((id) => {
|
||||
let unRegMsg = {messageType: "unregister",
|
||||
channelID: id};
|
||||
this._pushSocket.send(unRegMsg);
|
||||
});
|
||||
this.registeredChannels = {};
|
||||
}
|
||||
|
||||
this.connectionState = CONNECTION_STATE_CLOSED;
|
||||
this.serviceState = SERVICE_STATE_OFFLINE;
|
||||
this._pushSocket.close();
|
||||
this._pushSocket = undefined;
|
||||
// NOTE: this PushSocket instance will not be released until at least
|
||||
// the websocket referencing it as an nsIWebSocketListener is released.
|
||||
this.channels.clear();
|
||||
this.uaID = undefined;
|
||||
this.pushUrl = undefined;
|
||||
this.pushServerUri = undefined;
|
||||
},
|
||||
|
||||
/**
|
||||
* Start registration of a PushServer notification channel.
|
||||
* connection, it will automatically say hello and register the channel
|
||||
|
|
@ -97,159 +460,263 @@ let MozLoopPushHandler = {
|
|||
* @param {String} channelID Channel ID to use in registration.
|
||||
*
|
||||
* @param {Function} onRegistered Callback to be called once we are
|
||||
* registered.
|
||||
* registered.
|
||||
* NOTE: This function can be called multiple times if
|
||||
* the PushServer generates new pushURLs due to
|
||||
* re-registration due to network loss or PushServer
|
||||
* initiated re-assignment.
|
||||
* @param {Function} onNotification Callback to be called when a
|
||||
* push notification is received (may be called multiple
|
||||
* times).
|
||||
* push notification is received (may be called multiple
|
||||
* times).
|
||||
*/
|
||||
register: function(channelID, onRegistered, onNotification) {
|
||||
if (!channelID || !onRegistered || !onNotification) {
|
||||
throw new Error("missing required parameter(s):"
|
||||
+ (channelID ? "" : " channelID")
|
||||
+ (onRegistered ? "" : " onRegistered")
|
||||
+ (onNotification ? "" : " onNotification"));
|
||||
throw new Error("missing required parameter(s):" +
|
||||
(channelID ? "" : " channelID") +
|
||||
(onRegistered ? "" : " onRegistered") +
|
||||
(onNotification ? "" : " onNotification"));
|
||||
}
|
||||
|
||||
consoleLog.info("PushHandler: channel registration: ", channelID);
|
||||
// If the channel is already registered, callback with an error immediately
|
||||
// so we don't leave code hanging waiting for an onRegistered callback.
|
||||
if (channelID in this.channels) {
|
||||
if (this.channels.has(channelID)) {
|
||||
consoleLog.error("PushHandler: channel already registered");
|
||||
onRegistered("error: channel already registered: " + channelID);
|
||||
return;
|
||||
}
|
||||
|
||||
this.channels[channelID] = {
|
||||
onRegistered: onRegistered,
|
||||
onNotification: onNotification
|
||||
};
|
||||
this.channels.set(channelID, {onRegistered: onRegistered,
|
||||
onNotification: onNotification});
|
||||
|
||||
// If registration is in progress, simply add to the work list.
|
||||
// Else, re-start a registration cycle.
|
||||
if (this._registrationID) {
|
||||
this._channelsToRegister.push(channelID);
|
||||
} else {
|
||||
this._registerChannels();
|
||||
}
|
||||
this._channelsToRegister.push(channelID);
|
||||
this._registerChannels();
|
||||
},
|
||||
|
||||
/**
|
||||
* Listener method, handles the start of the websocket stream.
|
||||
* Handles the start of the websocket stream.
|
||||
* Sends a hello message to the server.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
*/
|
||||
onStart: function() {
|
||||
this._retryEnd();
|
||||
// If a uaID has already been assigned, assume this is a re-connect
|
||||
// and send the uaID in order to re-synch with the
|
||||
// PushServer. If a registration has been completed, send the channelID.
|
||||
_onStart: function() {
|
||||
consoleLog.info("PushHandler: websocket open, sending 'hello' to PushServer");
|
||||
this.connectionState = CONNECTION_STATE_OPEN;
|
||||
// If a uaID has already been assigned, assume this is a re-connect;
|
||||
// send the uaID and channelIDs in order to re-synch with the
|
||||
// PushServer. The PushServer does not need to accept the existing channelIDs
|
||||
// and may issue new channelIDs along with new pushURLs.
|
||||
this.serviceState = SERVICE_STATE_PENDING;
|
||||
let helloMsg = {
|
||||
messageType: "hello",
|
||||
uaid: this.uaID || "",
|
||||
channelIDs: Object.keys(this.registeredChannels)};
|
||||
|
||||
this._retryOperation(() => this.onStart(), this._maxRetryDelay_ms);
|
||||
try { // in case websocket has closed before this handler is run
|
||||
this._websocket.sendMsg(JSON.stringify(helloMsg));
|
||||
}
|
||||
catch (e) {console.warn("MozLoopPushHandler::onStart websocket.sendMsg() failure");}
|
||||
messageType: "hello",
|
||||
uaid: this.uaID || "",
|
||||
channelIDs: this.uaID ? Object.keys(this.registeredChannels) : []
|
||||
};
|
||||
// The Simple PushServer spec does not allow a retry of the Hello handshake but requires that the socket
|
||||
// be closed and another socket openned in order to re-attempt the handshake.
|
||||
// Here, the retryManager is not set up to retry the sending another 'hello' message: the timeout will
|
||||
// trigger closing the websocket and starting the connection again from the start.
|
||||
this._retryManager.reset();
|
||||
this._retryManager.retry(() => this._restartConnection());
|
||||
this._pushSocket.send(helloMsg);
|
||||
},
|
||||
|
||||
/**
|
||||
* Listener method, called when the websocket is closed.
|
||||
* Handles websocket close callbacks.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {nsresult} aStatusCode Reason for stopping (NS_OK = successful)
|
||||
* This method will continually try to re-establish a connection
|
||||
* to the PushServer unless shutdown has been called.
|
||||
*/
|
||||
onStop: function(aContext, aStatusCode) {
|
||||
Cu.reportError("Loop Push server web socket closed! Code: " + aStatusCode);
|
||||
this._retryOperation(() => this._openSocket());
|
||||
},
|
||||
_onClose: function(aCode, aReason) {
|
||||
this._pingMonitor.stop();
|
||||
|
||||
/**
|
||||
* Listener method, called when the websocket is closed by the server.
|
||||
* If there are errors, onStop may be called without ever calling this
|
||||
* method.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {integer} aCode the websocket closing handshake close code
|
||||
* @param {String} aReason the websocket closing handshake close reason
|
||||
*/
|
||||
onServerClose: function(aContext, aCode) {
|
||||
Cu.reportError("Loop Push server web socket closed (server)! Code: " + aCode);
|
||||
this._retryOperation(() => this._openSocket());
|
||||
},
|
||||
switch (this.connectionState) {
|
||||
case CONNECTION_STATE_OPEN:
|
||||
this.connectionState = CONNECTION_STATE_CLOSED;
|
||||
consoleLog.info("PushHandler: websocket closed: begin reconnect - ", aCode);
|
||||
// The first retry is immediate
|
||||
this._retryManager.reset();
|
||||
this._openSocket();
|
||||
break;
|
||||
|
||||
case CONNECTION_STATE_CONNECTING:
|
||||
// Wait before re-attempting to open the websocket.
|
||||
consoleLog.info("PushHandler: websocket closed: delay and retry - ", aCode);
|
||||
this._retryManager.retry(() => this._openSocket());
|
||||
break;
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Listener method, called when the websocket receives a message.
|
||||
*
|
||||
* @param {nsISupports} aContext Not used
|
||||
* @param {String} aMsg The message data
|
||||
* @param {Object} aMsg The message data
|
||||
*/
|
||||
onMessageAvailable: function(aContext, aMsg) {
|
||||
let msg = JSON.parse(aMsg);
|
||||
_onMsg: function(aMsg) {
|
||||
// If an error property exists in the message object ignore the other
|
||||
// properties.
|
||||
if (aMsg.error) {
|
||||
consoleLog.error("PushHandler: received error response msg: ", aMsg.error);
|
||||
return;
|
||||
}
|
||||
|
||||
switch(msg.messageType) {
|
||||
// The recommended response to a ping message when the push server has nothing
|
||||
// else to send is a blank JSON message body: {}
|
||||
if (!aMsg.messageType && this.serviceState === SERVICE_STATE_ACTIVE) {
|
||||
// Treat this as a ping response
|
||||
this._pingMonitor.restart();
|
||||
return;
|
||||
}
|
||||
|
||||
switch(aMsg.messageType) {
|
||||
case "hello":
|
||||
this._retryEnd();
|
||||
this._isConnected = true;
|
||||
if (this.uaID !== msg.uaid) {
|
||||
this.uaID = msg.uaid;
|
||||
this.registeredChannels = {};
|
||||
this._registerChannels();
|
||||
}
|
||||
this._onHello(aMsg);
|
||||
break;
|
||||
|
||||
case "register":
|
||||
this._onRegister(msg);
|
||||
this._onRegister(aMsg);
|
||||
break;
|
||||
|
||||
case "notification":
|
||||
msg.updates.forEach((update) => {
|
||||
if (update.channelID in this.registeredChannels) {
|
||||
this.channels[update.channelID].onNotification(update.version, update.channelID);
|
||||
}
|
||||
});
|
||||
this._onNotification(aMsg);
|
||||
break;
|
||||
|
||||
default:
|
||||
consoleLog.warn("PushHandler: unknown message type = ", aMsg.messageType);
|
||||
if (this.serviceState === SERVICE_STATE_ACTIVE) {
|
||||
// Treat this as a ping response
|
||||
this._pingMonitor.restart();
|
||||
}
|
||||
break;
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Handles hello message.
|
||||
*
|
||||
* This method will parse the hello response from the PushServer
|
||||
* and determine whether registration is necessary.
|
||||
*
|
||||
* @param {aMsg} hello message body
|
||||
*/
|
||||
_onHello: function(aMsg) {
|
||||
if (this.serviceState !== SERVICE_STATE_PENDING) {
|
||||
consoleLog.error("PushHandler: extra 'hello' response received from PushServer");
|
||||
return;
|
||||
}
|
||||
|
||||
// Clear any pending timeout that will restart the connection.
|
||||
this._retryManager.reset();
|
||||
this.serviceState = SERVICE_STATE_ACTIVE;
|
||||
consoleLog.info("PushHandler: 'hello' handshake complete");
|
||||
// Start the PushServer ping monitor
|
||||
this._pingMonitor.restart();
|
||||
// If a new uaID is received, then any previous channel registrations
|
||||
// are no longer valid and a Registration request is generated.
|
||||
if (this.uaID !== aMsg.uaid) {
|
||||
consoleLog.log("PushHandler: registering all channels");
|
||||
this.uaID = aMsg.uaid;
|
||||
// Re-register all channels.
|
||||
this._channelsToRegister = [...this.channels.keys()];
|
||||
this.registeredChannels = {};
|
||||
}
|
||||
// Allow queued registrations to start (or all if cleared above).
|
||||
this._registerChannels();
|
||||
},
|
||||
|
||||
/**
|
||||
* Handles notification message.
|
||||
*
|
||||
* This method will parse the Array of updates and trigger
|
||||
* the callback of any registered channel.
|
||||
* This method will construct an ack message containing
|
||||
* a set of channel version update notifications.
|
||||
*
|
||||
* @param {aMsg} notification message body
|
||||
*/
|
||||
_onNotification: function(aMsg) {
|
||||
if (this.serviceState !== SERVICE_STATE_ACTIVE ||
|
||||
this.registeredChannels.length === 0) {
|
||||
// Treat reception of a notification before handshake and registration
|
||||
// are complete as a fatal error.
|
||||
consoleLog.error("PushHandler: protocol error - notification received in wrong state");
|
||||
this._restartConnection();
|
||||
return;
|
||||
}
|
||||
|
||||
this._pingMonitor.restart();
|
||||
if (Array.isArray(aMsg.updates) && aMsg.updates.length > 0) {
|
||||
let ackChannels = [];
|
||||
aMsg.updates.forEach(update => {
|
||||
if (update.channelID in this.registeredChannels) {
|
||||
consoleLog.log("PushHandler: notification: version = ", update.version,
|
||||
", channelID = ", update.channelID);
|
||||
this.channels.get(update.channelID)
|
||||
.onNotification(update.version, update.channelID);
|
||||
ackChannels.push(update);
|
||||
} else {
|
||||
consoleLog.error("PushHandler: notification received for unknown channelID: ",
|
||||
update.channelID);
|
||||
}
|
||||
});
|
||||
|
||||
consoleLog.log("PushHandler: PusherServer 'ack': ", ackChannels);
|
||||
this._pushSocket.send({messageType: "ack",
|
||||
updates: ackChannels});
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Handles the PushServer registration response.
|
||||
*
|
||||
* @param {Object} msg PushServer to UserAgent registration response (parsed from JSON).
|
||||
*/
|
||||
_onRegister: function(msg) {
|
||||
let registerNext = () => {
|
||||
this._registrationID = this._channelsToRegister.shift();
|
||||
this._sendRegistration(this._registrationID);
|
||||
if (this.serviceState !== SERVICE_STATE_ACTIVE ||
|
||||
msg.channelID != this._pendingChannelID) {
|
||||
// Treat reception of a register response outside of a completed handshake
|
||||
// or for a channelID not currently pending a response
|
||||
// as an indication that the connections should be reset.
|
||||
consoleLog.error("PushHandler: registration protocol error");
|
||||
this._restartConnection();
|
||||
return;
|
||||
}
|
||||
|
||||
this._retryManager.reset();
|
||||
this._pingMonitor.restart();
|
||||
|
||||
switch (msg.status) {
|
||||
case 200:
|
||||
if (msg.channelID == this._registrationID) {
|
||||
this._retryEnd(); // reset retry mechanism
|
||||
this.registeredChannels[msg.channelID] = msg.pushEndpoint;
|
||||
this.channels[msg.channelID].onRegistered(null, msg.pushEndpoint, msg.channelID);
|
||||
registerNext();
|
||||
}
|
||||
consoleLog.info("PushHandler: channel registered: ", msg.channelID);
|
||||
this.registeredChannels[msg.channelID] = msg.pushEndpoint;
|
||||
this.channels.get(msg.channelID)
|
||||
.onRegistered(null, msg.pushEndpoint, msg.channelID);
|
||||
this._registerNext();
|
||||
break;
|
||||
|
||||
case 500:
|
||||
consoleLog.info("PushHandler: eeceived a 500 retry response from the PushServer: ",
|
||||
msg.channelID);
|
||||
// retry the registration request after a suitable delay
|
||||
this._retryOperation(() => this._sendRegistration(msg.channelID));
|
||||
this._retryManager.retry(() => this._sendRegistration(msg.channelID));
|
||||
break;
|
||||
|
||||
case 409:
|
||||
this.channels[this._registrationID].onRegistered(
|
||||
consoleLog.error("PushHandler: received a 409 response from the PushServer: ",
|
||||
msg.channelID);
|
||||
this.channels.get(this._pendingChannelID).onRegistered(
|
||||
"error: PushServer ChannelID already in use: " + msg.channelID);
|
||||
registerNext();
|
||||
// Remove this channel from the channel list.
|
||||
this.channels.delete(this._pendingChannelID);
|
||||
this._registerNext();
|
||||
break;
|
||||
|
||||
default:
|
||||
let id = this._channelsToRegister.shift();
|
||||
this.channels[this._registrationID].onRegistered(
|
||||
consoleLog.error("PushHandler: received error ", msg.status,
|
||||
" from the PushServer: ", msg.channelID);
|
||||
this.channels.get(this._pendingChannelID).onRegistered(
|
||||
"error: PushServer registration failure, status = " + msg.status);
|
||||
registerNext();
|
||||
this.channels.delete(this._pendingChannelID);
|
||||
this._registerNext();
|
||||
break;
|
||||
}
|
||||
},
|
||||
|
|
@ -259,38 +726,39 @@ let MozLoopPushHandler = {
|
|||
*
|
||||
* A new websocket interface is used each time. If an onStop callback
|
||||
* was received, calling asyncOpen() on the same interface will
|
||||
* trigger a "alreay open socket" exception even though the channel
|
||||
* trigger an "already open socket" exception even though the channel
|
||||
* is logically closed.
|
||||
*/
|
||||
_openSocket: function() {
|
||||
this._isConnected = false;
|
||||
|
||||
if (this._mockWebSocket) {
|
||||
// For tests, use the mock instance.
|
||||
this._websocket = this._mockWebSocket;
|
||||
} else {
|
||||
this._websocket = Cc["@mozilla.org/network/protocol;1?name=wss"]
|
||||
.createInstance(Ci.nsIWebSocketChannel);
|
||||
}
|
||||
|
||||
this._websocket.protocol = "push-notification";
|
||||
this.connectionState = CONNECTION_STATE_CONNECTING;
|
||||
// For tests, use the mock instance.
|
||||
this._pushSocket = new PushSocket(this._mockWebSocket);
|
||||
|
||||
let performOpen = () => {
|
||||
let uri = Services.io.newURI(this.pushServerUri, null, null);
|
||||
this._websocket.asyncOpen(uri, this.pushServerUri, this, null);
|
||||
consoleLog.info("PushHandler: attempt to open websocket to PushServer: ", this.pushServerUri);
|
||||
this._pushSocket.connect(this.pushServerUri,
|
||||
(aMsg) => this._onMsg(aMsg),
|
||||
() => this._onStart(),
|
||||
(aCode, aReason) => this._onClose(aCode, aReason));
|
||||
}
|
||||
|
||||
let pushServerURLFetchError = () => {
|
||||
console.warn("MozLoopPushHandler - Could not retrieve push server URL from Loop server, will retry");
|
||||
this._retryOperation(() => this._openSocket());
|
||||
consoleLog.warn("PushHandler: Could not retrieve push server URL from Loop server, will retry");
|
||||
this._pushSocket = undefined;
|
||||
this._retryManager.retry(() => this._openSocket());
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
this.pushServerUri = Services.prefs.getCharPref("loop.debug.pushserver");
|
||||
}
|
||||
catch (e) {}
|
||||
|
||||
if (!this.pushServerUri) {
|
||||
// Get push server to use from the Loop server
|
||||
let pushUrlEndpoint = Services.prefs.getCharPref("loop.server") + "/push-server-config";
|
||||
let req = Cc["@mozilla.org/xmlextras/xmlhttprequest;1"].
|
||||
createInstance(Ci.nsIXMLHttpRequest);
|
||||
createInstance(Ci.nsIXMLHttpRequest);
|
||||
req.open("GET", pushUrlEndpoint);
|
||||
req.onload = () => {
|
||||
if (req.status >= 200 && req.status < 300) {
|
||||
|
|
@ -298,19 +766,19 @@ let MozLoopPushHandler = {
|
|||
try {
|
||||
pushServerConfig = JSON.parse(req.responseText);
|
||||
} catch (e) {
|
||||
console.warn("MozLoopPushHandler - Error parsing JSON response for push server URL");
|
||||
consoleLog.warn("PushHandler: Error parsing JSON response for push server URL");
|
||||
pushServerURLFetchError();
|
||||
}
|
||||
if (pushServerConfig.pushServerURI) {
|
||||
this._retryManager.reset();
|
||||
this.pushServerUri = pushServerConfig.pushServerURI;
|
||||
this._retryEnd();
|
||||
performOpen();
|
||||
} else {
|
||||
console.warn("MozLoopPushHandler - push server URL config lacks pushServerURI parameter");
|
||||
consoleLog.warn("PushHandler: push server URL config lacks pushServerURI parameter");
|
||||
pushServerURLFetchError();
|
||||
}
|
||||
} else {
|
||||
console.warn("MozLoopPushHandler - push server URL retrieve error: " + req.status);
|
||||
consoleLog.warn("PushHandler: push server URL retrieve error: " + req.status);
|
||||
pushServerURLFetchError();
|
||||
}
|
||||
};
|
||||
|
|
@ -322,25 +790,43 @@ let MozLoopPushHandler = {
|
|||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Closes websocket and begins re-establishing a connection with the PushServer
|
||||
*/
|
||||
_restartConnection: function() {
|
||||
this._retryManager.reset();
|
||||
this._pingMonitor.stop();
|
||||
this.serviceState = SERVICE_STATE_OFFLINE;
|
||||
this._pendingChannelID = null;
|
||||
|
||||
if (this.connectionState === CONNECTION_STATE_OPEN) {
|
||||
// Close the current PushSocket and start the operation to open a new one.
|
||||
this.connectionState = CONNECTION_STATE_CLOSED;
|
||||
this._pushSocket.close();
|
||||
consoleLog.warn("PushHandler: connection error: re-establishing connection to PushServer");
|
||||
this._openSocket();
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Begins registering the channelIDs with the PushServer
|
||||
*/
|
||||
_registerChannels: function() {
|
||||
// Hold off registration operation until handshake is complete.
|
||||
if (!this._isConnected) {
|
||||
// If a registration cycle is in progress, do nothing.
|
||||
if (this.serviceState !== SERVICE_STATE_ACTIVE ||
|
||||
this._pendingChannelID) {
|
||||
return;
|
||||
}
|
||||
this._registerNext();
|
||||
},
|
||||
|
||||
// If a registration is pending, do not generate a work list.
|
||||
// Assume registration is in progress.
|
||||
if (!this._registrationID) {
|
||||
// Generate a list of channelIDs that have not yet been registered.
|
||||
this._channelsToRegister = Object.keys(this.channels).filter((id) => {
|
||||
return !(id in this.registeredChannels);
|
||||
});
|
||||
this._registrationID = this._channelsToRegister.shift();
|
||||
this._sendRegistration(this._registrationID);
|
||||
}
|
||||
/**
|
||||
* Gets the next channel to register from the worklist and kicks of its registration
|
||||
*/
|
||||
_registerNext: function() {
|
||||
this._pendingChannelID = this._channelsToRegister.pop();
|
||||
this._sendRegistration(this._pendingChannelID);
|
||||
},
|
||||
|
||||
/**
|
||||
|
|
@ -350,42 +836,8 @@ let MozLoopPushHandler = {
|
|||
*/
|
||||
_sendRegistration: function(channelID) {
|
||||
if (channelID) {
|
||||
try { // in case websocket has closed
|
||||
this._websocket.sendMsg(JSON.stringify({messageType: "register",
|
||||
channelID: channelID}));
|
||||
}
|
||||
catch (e) {console.warn("MozLoopPushHandler::_registerChannel websocket.sendMsg() failure");}
|
||||
this._pushSocket.send({messageType: "register",
|
||||
channelID: channelID});
|
||||
}
|
||||
},
|
||||
|
||||
/**
|
||||
* Method to handle retrying UserAgent to PushServer request following
|
||||
* a retry back-off scheme managed by this function.
|
||||
*
|
||||
* @param {function} delayedOp Function to call after current delay is satisfied
|
||||
*
|
||||
* @param {number} [optional] retryDelay This parameter will be used as the initial delay
|
||||
*/
|
||||
_retryOperation: function(delayedOp, retryDelay) {
|
||||
if (!this._retryCount) {
|
||||
this._retryDelay = retryDelay || this._minRetryDelay_ms;
|
||||
this._retryCount = 1;
|
||||
} else {
|
||||
let nextDelay = this._retryDelay * 2;
|
||||
this._retryDelay = nextDelay > this._maxRetryDelay_ms ? this._maxRetryDelay_ms : nextDelay;
|
||||
this._retryCount += 1;
|
||||
}
|
||||
this._timeoutID = setTimeout(delayedOp, this._retryDelay);
|
||||
},
|
||||
|
||||
/**
|
||||
* Method used to reset the retry delay back-off logic.
|
||||
*
|
||||
*/
|
||||
_retryEnd: function() {
|
||||
if (this._retryCount) {
|
||||
clearTimeout(this._timeoutID);
|
||||
this._retryCount = 0;
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче