gecko-dev/dom/push/PushService.jsm
William Chen 8588891ca7 Bug 1210211 - Part 1: Delay updating push quota. r=kitcambridge
--HG--
extra : commitid : 70QSm4u4WLW
extra : rebase_source : e52cc3bae822448e64dca6c40be48d74ea924e0b
2015-11-16 21:33:11 -08:00

1459 lines
48 KiB
JavaScript

/* jshint moz: true, esnext: true */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this file,
* You can obtain one at http://mozilla.org/MPL/2.0/. */
"use strict";
const Cc = Components.classes;
const Ci = Components.interfaces;
const Cu = Components.utils;
const Cr = Components.results;
const {PushDB} = Cu.import("resource://gre/modules/PushDB.jsm");
Cu.import("resource://gre/modules/XPCOMUtils.jsm");
Cu.import("resource://gre/modules/Services.jsm");
Cu.import("resource://gre/modules/Timer.jsm");
Cu.import("resource://gre/modules/Preferences.jsm");
Cu.import("resource://gre/modules/Promise.jsm");
const {PushServiceWebSocket} = Cu.import("resource://gre/modules/PushServiceWebSocket.jsm");
const {PushServiceHttp2} = Cu.import("resource://gre/modules/PushServiceHttp2.jsm");
const {PushCrypto} = Cu.import("resource://gre/modules/PushCrypto.jsm");
// Currently supported protocols: WebSocket.
const CONNECTION_PROTOCOLS = [PushServiceWebSocket, PushServiceHttp2];
XPCOMUtils.defineLazyModuleGetter(this, "AlarmService",
"resource://gre/modules/AlarmService.jsm");
XPCOMUtils.defineLazyServiceGetter(this, "gContentSecurityManager",
"@mozilla.org/contentsecuritymanager;1",
"nsIContentSecurityManager");
this.EXPORTED_SYMBOLS = ["PushService"];
XPCOMUtils.defineLazyGetter(this, "console", () => {
let {ConsoleAPI} = Cu.import("resource://gre/modules/Console.jsm", {});
return new ConsoleAPI({
maxLogLevelPref: "dom.push.loglevel",
prefix: "PushService",
});
});
const prefs = new Preferences("dom.push.");
const kCHILD_PROCESS_MESSAGES = ["Push:Register", "Push:Unregister",
"Push:Registration", "Push:RegisterEventNotificationListener",
"Push:NotificationForOriginShown",
"Push:NotificationForOriginClosed",
"child-process-shutdown"];
const PUSH_SERVICE_UNINIT = 0;
const PUSH_SERVICE_INIT = 1; // No serverURI
const PUSH_SERVICE_ACTIVATING = 2;//activating db
const PUSH_SERVICE_CONNECTION_DISABLE = 3;
const PUSH_SERVICE_ACTIVE_OFFLINE = 4;
const PUSH_SERVICE_RUNNING = 5;
// Telemetry failure to send push notification to Service Worker reasons.
// Key not found in local database.
const kDROP_NOTIFICATION_REASON_KEY_NOT_FOUND = 0;
// User cleared history.
const kDROP_NOTIFICATION_REASON_NO_HISTORY = 1;
// Version of message received not newer than previous one.
const kDROP_NOTIFICATION_REASON_NO_VERSION_INCREMENT = 2;
// Subscription has expired.
const kDROP_NOTIFICATION_REASON_EXPIRED = 3;
/**
* State is change only in couple of functions:
* init - change state to PUSH_SERVICE_INIT if state was PUSH_SERVICE_UNINIT
* changeServerURL - change state to PUSH_SERVICE_ACTIVATING if serverURL
* present or PUSH_SERVICE_INIT if not present.
* changeStateConnectionEnabledEvent - it is call on pref change or during
* the service activation and it can
* change state to
* PUSH_SERVICE_CONNECTION_DISABLE
* changeStateOfflineEvent - it is called when offline state changes or during
* the service activation and it change state to
* PUSH_SERVICE_ACTIVE_OFFLINE or
* PUSH_SERVICE_RUNNING.
* uninit - change state to PUSH_SERVICE_UNINIT.
**/
// This is for starting and stopping service.
const STARTING_SERVICE_EVENT = 0;
const CHANGING_SERVICE_EVENT = 1;
const STOPPING_SERVICE_EVENT = 2;
const UNINIT_EVENT = 3;
/**
* The implementation of the push system. It uses WebSockets
* (PushServiceWebSocket) to communicate with the server and PushDB (IndexedDB)
* for persistence.
*/
this.PushService = {
_service: null,
_state: PUSH_SERVICE_UNINIT,
_db: null,
_options: null,
_alarmID: null,
_visibleNotifications: new Map(),
// Callback that is called after attempting to
// reduce the quota for a record. Used for testing purposes.
_updateQuotaTestCallback: null,
_childListeners: [],
// When serverURI changes (this is used for testing), db is cleaned up and a
// a new db is started. This events must be sequential.
_stateChangeProcessQueue: null,
_stateChangeProcessEnqueue: function(op) {
if (!this._stateChangeProcessQueue) {
this._stateChangeProcessQueue = Promise.resolve();
}
this._stateChangeProcessQueue = this._stateChangeProcessQueue
.then(op)
.catch(_ => {});
},
// Pending request. If a worker try to register for the same scope again, do
// not send a new registration request. Therefore we need queue of pending
// register requests. This is the list of scopes which pending registration.
_pendingRegisterRequest: {},
_notifyActivated: null,
_activated: null,
_checkActivated: function() {
if (this._state < PUSH_SERVICE_ACTIVATING) {
return Promise.reject(new Error("Push service not active"));
} else if (this._state > PUSH_SERVICE_ACTIVATING) {
return Promise.resolve();
} else {
return (this._activated) ? this._activated :
this._activated = new Promise((res, rej) =>
this._notifyActivated = {resolve: res,
reject: rej});
}
},
_makePendingKey: function(aPageRecord) {
return aPageRecord.scope + "|" + aPageRecord.originAttributes;
},
_lookupOrPutPendingRequest: function(aPageRecord) {
let key = this._makePendingKey(aPageRecord);
if (this._pendingRegisterRequest[key]) {
return this._pendingRegisterRequest[key];
}
return this._pendingRegisterRequest[key] = this._registerWithServer(aPageRecord);
},
_deletePendingRequest: function(aPageRecord) {
let key = this._makePendingKey(aPageRecord);
if (this._pendingRegisterRequest[key]) {
delete this._pendingRegisterRequest[key];
}
},
_setState: function(aNewState) {
console.debug("setState()", "new state", aNewState, "old state", this._state);
if (this._state == aNewState) {
return;
}
if (this._state == PUSH_SERVICE_ACTIVATING) {
// It is not important what is the new state as soon as we leave
// PUSH_SERVICE_ACTIVATING
this._state = aNewState;
if (this._notifyActivated) {
if (aNewState < PUSH_SERVICE_ACTIVATING) {
this._notifyActivated.reject(new Error("Push service not active"));
} else {
this._notifyActivated.resolve();
}
}
this._notifyActivated = null;
this._activated = null;
}
this._state = aNewState;
},
_changeStateOfflineEvent: function(offline, calledFromConnEnabledEvent) {
console.debug("changeStateOfflineEvent()", offline);
if (this._state < PUSH_SERVICE_ACTIVE_OFFLINE &&
this._state != PUSH_SERVICE_ACTIVATING &&
!calledFromConnEnabledEvent) {
return;
}
if (offline) {
if (this._state == PUSH_SERVICE_RUNNING) {
this._service.disconnect();
}
this._setState(PUSH_SERVICE_ACTIVE_OFFLINE);
} else {
if (this._state == PUSH_SERVICE_RUNNING) {
// PushService was not in the offline state, but got notification to
// go online (a offline notification has not been sent).
// Disconnect first.
this._service.disconnect();
}
this._db.getAllUnexpired()
.then(records => {
if (records.length > 0) {
// if there are request waiting
this._service.connect(records);
}
});
this._setState(PUSH_SERVICE_RUNNING);
}
},
_changeStateConnectionEnabledEvent: function(enabled) {
console.debug("changeStateConnectionEnabledEvent()", enabled);
if (this._state < PUSH_SERVICE_CONNECTION_DISABLE &&
this._state != PUSH_SERVICE_ACTIVATING) {
return;
}
if (enabled) {
this._changeStateOfflineEvent(Services.io.offline, true);
} else {
if (this._state == PUSH_SERVICE_RUNNING) {
this._service.disconnect();
}
this._setState(PUSH_SERVICE_CONNECTION_DISABLE);
}
},
observe: function observe(aSubject, aTopic, aData) {
switch (aTopic) {
/*
* We need to call uninit() on shutdown to clean up things that modules
* aren't very good at automatically cleaning up, so we don't get shutdown
* leaks on browser shutdown.
*/
case "xpcom-shutdown":
this.uninit();
break;
case "network-active-changed": /* On B2G. */
case "network:offline-status-changed": /* On desktop. */
this._stateChangeProcessEnqueue(_ =>
this._changeStateOfflineEvent(aData === "offline", false)
);
break;
case "nsPref:changed":
if (aData == "dom.push.serverURL") {
console.debug("observe: dom.push.serverURL changed for websocket",
prefs.get("serverURL"));
this._stateChangeProcessEnqueue(_ =>
this._changeServerURL(prefs.get("serverURL"),
CHANGING_SERVICE_EVENT)
);
} else if (aData == "dom.push.connection.enabled") {
this._stateChangeProcessEnqueue(_ =>
this._changeStateConnectionEnabledEvent(prefs.get("connection.enabled"))
);
}
break;
case "idle-daily":
this._dropExpiredRegistrations();
break;
case "perm-changed":
this._onPermissionChange(aSubject, aData).catch(error => {
console.error("onPermissionChange: Error updating registrations:",
error);
})
break;
case "clear-origin-data":
this._clearOriginData(data).catch(error => {
console.error("clearOriginData: Error clearing origin data:", error);
});
break;
}
},
_clearOriginData: function(data) {
console.log("clearOriginData()");
if (!data) {
return Promise.resolve();
}
let pattern = JSON.parse(data);
return this._db.clearIf(record => {
if (!record.matchesOriginAttributes(pattern)) {
return false;
}
this._backgroundUnregister(record);
return true;
});
},
/**
* Sends an unregister request to the server in the background. If the
* service is not connected, this function is a no-op.
*
* @param {PushRecord} record The record to unregister.
*/
_backgroundUnregister: function(record) {
console.debug("backgroundUnregister()");
if (!this._service.isConnected() || !record) {
return;
}
console.debug("backgroundUnregister: Notifying server", record);
this._sendUnregister(record).catch(e => {
console.error("backgroundUnregister: Error notifying server", e);
});
},
// utility function used to add/remove observers in startObservers() and
// stopObservers()
getNetworkStateChangeEventName: function() {
try {
Cc["@mozilla.org/network/manager;1"].getService(Ci.nsINetworkManager);
return "network-active-changed";
} catch (e) {
return "network:offline-status-changed";
}
},
_findService: function(serverURL) {
console.debug("findService()");
let uri;
let service;
if (!serverURL) {
console.warn("findService: No dom.push.serverURL found");
return [];
}
try {
uri = Services.io.newURI(serverURL, null, null);
} catch (e) {
console.warn("findService: Error creating valid URI from",
"dom.push.serverURL", serverURL);
return [];
}
if (!gContentSecurityManager.isURIPotentiallyTrustworthy(uri)) {
console.warn("findService: Untrusted server URI", uri.spec);
return [];
}
for (let connProtocol of CONNECTION_PROTOCOLS) {
if (connProtocol.validServerURI(uri)) {
service = connProtocol;
break;
}
}
return [service, uri];
},
_changeServerURL: function(serverURI, event) {
console.debug("changeServerURL()");
switch(event) {
case UNINIT_EVENT:
return this._stopService(event);
case STARTING_SERVICE_EVENT:
{
let [service, uri] = this._findService(serverURI);
if (!service) {
this._setState(PUSH_SERVICE_INIT);
return Promise.resolve();
}
return this._startService(service, uri, event)
.then(_ => this._stateChangeProcessEnqueue(_ =>
this._changeStateConnectionEnabledEvent(prefs.get("connection.enabled")))
);
}
case CHANGING_SERVICE_EVENT:
let [service, uri] = this._findService(serverURI);
if (service) {
if (this._state == PUSH_SERVICE_INIT) {
this._setState(PUSH_SERVICE_ACTIVATING);
// The service has not been running - start it.
return this._startService(service, uri, STARTING_SERVICE_EVENT)
.then(_ => this._stateChangeProcessEnqueue(_ =>
this._changeStateConnectionEnabledEvent(prefs.get("connection.enabled")))
);
} else {
this._setState(PUSH_SERVICE_ACTIVATING);
// If we already had running service - stop service, start the new
// one and check connection.enabled and offline state(offline state
// check is called in changeStateConnectionEnabledEvent function)
return this._stopService(CHANGING_SERVICE_EVENT)
.then(_ =>
this._startService(service, uri, CHANGING_SERVICE_EVENT)
)
.then(_ => this._stateChangeProcessEnqueue(_ =>
this._changeStateConnectionEnabledEvent(prefs.get("connection.enabled")))
);
}
} else {
if (this._state == PUSH_SERVICE_INIT) {
return Promise.resolve();
} else {
// The new serverUri is empty or misconfigured - stop service.
this._setState(PUSH_SERVICE_INIT);
return this._stopService(STOPPING_SERVICE_EVENT);
}
}
}
},
/**
* PushService initialization is divided into 4 parts:
* init() - start listening for xpcom-shutdown and serverURL changes.
* state is change to PUSH_SERVICE_INIT
* startService() - if serverURL is present this function is called. It starts
* listening for broadcasted messages, starts db and
* PushService connection (WebSocket).
* state is change to PUSH_SERVICE_ACTIVATING.
* startObservers() - start other observers.
* changeStateConnectionEnabledEvent - checks prefs and offline state.
* It changes state to:
* PUSH_SERVICE_RUNNING,
* PUSH_SERVICE_ACTIVE_OFFLINE or
* PUSH_SERVICE_CONNECTION_DISABLE.
*/
init: function(options = {}) {
console.debug("init()");
if (this._state > PUSH_SERVICE_UNINIT) {
return;
}
this._setState(PUSH_SERVICE_ACTIVATING);
Services.obs.addObserver(this, "xpcom-shutdown", false);
if (options.serverURI) {
// this is use for xpcshell test.
let [service, uri] = this._findService(options.serverURI);
if (!service) {
this._setState(PUSH_SERVICE_INIT);
return;
}
// Start service.
this._startService(service, uri, false, options).then(_ => {
// Before completing the activation check prefs. This will first check
// connection.enabled pref and then check offline state.
this._changeStateConnectionEnabledEvent(prefs.get("connection.enabled"));
});
} else {
// This is only used for testing. Different tests require connecting to
// slightly different URLs.
prefs.observe("serverURL", this);
this._stateChangeProcessEnqueue(_ =>
this._changeServerURL(prefs.get("serverURL"), STARTING_SERVICE_EVENT));
}
},
_startObservers: function() {
console.debug("startObservers()");
if (this._state != PUSH_SERVICE_ACTIVATING) {
return;
}
Services.obs.addObserver(this, "clear-origin-data", false);
// On B2G the NetworkManager interface fires a network-active-changed
// event.
//
// The "active network" is based on priority - i.e. Wi-Fi has higher
// priority than data. The PushService should just use the preferred
// network, and not care about all interface changes.
// network-active-changed is not fired when the network goes offline, but
// socket connections time out. The check for Services.io.offline in
// PushServiceWebSocket._beginWSSetup() prevents unnecessary retries. When
// the network comes back online, network-active-changed is fired.
//
// On non-B2G platforms, the offline-status-changed event is used to know
// when to (dis)connect. It may not fire if the underlying OS changes
// networks; in such a case we rely on timeout.
//
// On B2G both events fire, one after the other, when the network goes
// online, so we explicitly check for the presence of NetworkManager and
// don't add an observer for offline-status-changed on B2G.
this._networkStateChangeEventName = this.getNetworkStateChangeEventName();
Services.obs.addObserver(this, this._networkStateChangeEventName, false);
// Used to monitor if the user wishes to disable Push.
prefs.observe("connection.enabled", this);
// Prunes expired registrations and notifies dormant service workers.
Services.obs.addObserver(this, "idle-daily", false);
// Prunes registrations for sites for which the user revokes push
// permissions.
Services.obs.addObserver(this, "perm-changed", false);
},
_startService: function(service, serverURI, event, options = {}) {
console.debug("startService()");
if (this._state != PUSH_SERVICE_ACTIVATING) {
return;
}
if (event != CHANGING_SERVICE_EVENT) {
// if only serverURL is changed we can keep listening for broadcast
// messages and queue them.
let ppmm = Cc["@mozilla.org/parentprocessmessagemanager;1"]
.getService(Ci.nsIMessageBroadcaster);
kCHILD_PROCESS_MESSAGES.forEach(msgName =>
ppmm.addMessageListener(msgName, this)
);
}
this._service = service;
this._db = options.db;
if (!this._db) {
this._db = this._service.newPushDB();
}
this._service.init(options, this, serverURI);
this._startObservers();
return this._dropExpiredRegistrations();
},
/**
* PushService uninitialization is divided into 3 parts:
* stopObservers() - stot observers started in startObservers.
* stopService() - It stops listening for broadcasted messages, stops db and
* PushService connection (WebSocket).
* state is changed to PUSH_SERVICE_INIT.
* uninit() - stop listening for xpcom-shutdown and serverURL changes.
* state is change to PUSH_SERVICE_UNINIT
*/
_stopService: function(event) {
console.debug("stopService()");
if (this._state < PUSH_SERVICE_ACTIVATING) {
return;
}
this.stopAlarm();
this._stopObservers();
if (event != CHANGING_SERVICE_EVENT) {
let ppmm = Cc["@mozilla.org/parentprocessmessagemanager;1"]
.getService(Ci.nsIMessageBroadcaster);
kCHILD_PROCESS_MESSAGES.forEach(
msgName => ppmm.removeMessageListener(msgName, this)
);
}
this._service.disconnect();
this._service.uninit();
this._service = null;
this.stopAlarm();
if (!this._db) {
return Promise.resolve();
}
if (event == UNINIT_EVENT) {
// If it is uninitialized just close db.
this._db.close();
this._db = null;
return Promise.resolve();
}
return this.dropUnexpiredRegistrations()
.then(_ => {
this._db.close();
this._db = null;
}, err => {
this._db.close();
this._db = null;
});
},
_stopObservers: function() {
console.debug("stopObservers()");
if (this._state < PUSH_SERVICE_ACTIVATING) {
return;
}
prefs.ignore("connection.enabled", this);
Services.obs.removeObserver(this, this._networkStateChangeEventName);
Services.obs.removeObserver(this, "clear-origin-data");
Services.obs.removeObserver(this, "idle-daily");
Services.obs.removeObserver(this, "perm-changed");
},
uninit: function() {
console.debug("uninit()");
this._childListeners = [];
if (this._state == PUSH_SERVICE_UNINIT) {
return;
}
this._setState(PUSH_SERVICE_UNINIT);
prefs.ignore("serverURL", this);
Services.obs.removeObserver(this, "xpcom-shutdown");
this._stateChangeProcessEnqueue(_ =>
this._changeServerURL("", UNINIT_EVENT));
console.debug("uninit: shutdown complete!");
},
/** |delay| should be in milliseconds. */
setAlarm: function(delay) {
// Bug 909270: Since calls to AlarmService.add() are async, calls must be
// 'queued' to ensure only one alarm is ever active.
if (this._settingAlarm) {
// onSuccess will handle the set. Overwriting the variable enforces the
// last-writer-wins semantics.
this._queuedAlarmDelay = delay;
this._waitingForAlarmSet = true;
return;
}
// Stop any existing alarm.
this.stopAlarm();
this._settingAlarm = true;
AlarmService.add(
{
date: new Date(Date.now() + delay),
ignoreTimezone: true
},
() => {
if (this._state > PUSH_SERVICE_ACTIVATING) {
this._service.onAlarmFired();
}
}, (alarmID) => {
this._alarmID = alarmID;
console.debug("setAlarm: Set alarm", delay, "in the future",
this._alarmID);
this._settingAlarm = false;
if (this._waitingForAlarmSet) {
this._waitingForAlarmSet = false;
this.setAlarm(this._queuedAlarmDelay);
}
}
);
},
stopAlarm: function() {
if (this._alarmID !== null) {
console.debug("stopAlarm: Stopped existing alarm", this._alarmID);
AlarmService.remove(this._alarmID);
this._alarmID = null;
}
},
/**
* Drops all active registrations and notifies the associated service
* workers. This function is called when the user switches Push servers,
* or when the server invalidates all existing registrations.
*
* We ignore expired registrations because they're already handled in other
* code paths. Registrations that expired after exceeding their quotas are
* evicted at startup, or on the next `idle-daily` event. Registrations that
* expired because the user revoked the notification permission are evicted
* once the permission is reinstated.
*/
dropUnexpiredRegistrations: function() {
let subscriptionChanges = [];
return this._db.clearIf(record => {
if (record.isExpired()) {
return false;
}
subscriptionChanges.push(record);
return true;
}).then(() => {
this.notifySubscriptionChanges(subscriptionChanges);
});
},
_notifySubscriptionChangeObservers: function(record) {
if (!record) {
return;
}
// Notify XPCOM observers.
Services.obs.notifyObservers(
null,
"push-subscription-change",
record.scope
);
let data = {
originAttributes: record.originAttributes,
scope: record.scope
};
Services.telemetry.getHistogramById("PUSH_API_NOTIFY_REGISTRATION_LOST").add();
this._notifyListeners('pushsubscriptionchange', data);
},
_notifyListeners: function(name, data) {
if (this._childListeners.length > 0) {
// Try to send messages to all listeners, but remove any that fail since
// the receiver is likely gone away.
for (var i = this._childListeners.length - 1; i >= 0; --i) {
try {
this._childListeners[i].sendAsyncMessage(name, data);
} catch(e) {
this._childListeners.splice(i, 1);
}
}
} else {
let ppmm = Cc['@mozilla.org/parentprocessmessagemanager;1']
.getService(Ci.nsIMessageListenerManager);
ppmm.broadcastAsyncMessage(name, data);
}
},
/**
* Drops a registration and notifies the associated service worker. If the
* registration does not exist, this function is a no-op.
*
* @param {String} keyID The registration ID to remove.
* @returns {Promise} Resolves once the worker has been notified.
*/
dropRegistrationAndNotifyApp: function(aKeyID) {
return this._db.delete(aKeyID)
.then(record => this._notifySubscriptionChangeObservers(record));
},
/**
* Replaces an existing registration and notifies the associated service
* worker.
*
* @param {String} aOldKey The registration ID to replace.
* @param {PushRecord} aNewRecord The new record.
* @returns {Promise} Resolves once the worker has been notified.
*/
updateRegistrationAndNotifyApp: function(aOldKey, aNewRecord) {
return this.updateRecordAndNotifyApp(aOldKey, _ => aNewRecord);
},
/**
* Updates a registration and notifies the associated service worker.
*
* @param {String} keyID The registration ID to update.
* @param {Function} updateFunc Returns the updated record.
* @returns {Promise} Resolves with the updated record once the worker
* has been notified.
*/
updateRecordAndNotifyApp: function(aKeyID, aUpdateFunc) {
return this._db.update(aKeyID, aUpdateFunc)
.then(record => {
this._notifySubscriptionChangeObservers(record);
return record;
});
},
notifySubscriptionChanges: function(records) {
records.forEach(record => {
this._notifySubscriptionChangeObservers(record);
});
},
ensureP256dhKey: function(record) {
if (record.p256dhPublicKey && record.p256dhPrivateKey) {
return Promise.resolve(record);
}
// We do not have a encryption key. so we need to generate it. This
// is only going to happen on db upgrade from version 4 to higher.
return PushCrypto.generateKeys()
.then(exportedKeys => {
return this.updateRecordAndNotifyApp(record.keyID, record => {
record.p256dhPublicKey = exportedKeys[0];
record.p256dhPrivateKey = exportedKeys[1];
return record;
});
}, error => {
return this.dropRegistrationAndNotifyApp(record.keyID).then(
() => Promise.reject(error));
});
},
_recordDidNotNotify: function(reason) {
Services.telemetry.
getHistogramById("PUSH_API_NOTIFICATION_RECEIVED_BUT_DID_NOT_NOTIFY").
add(reason);
},
/**
* Dispatches an incoming message to a service worker, recalculating the
* quota for the associated push registration. If the quota is exceeded,
* the registration and message will be dropped, and the worker will not
* be notified.
*
* @param {String} keyID The push registration ID.
* @param {String} message The message contents.
* @param {Object} cryptoParams The message encryption settings.
* @param {Function} updateFunc A function that receives the existing
* registration record as its argument, and returns a new record. If the
* function returns `null` or `undefined`, the record will not be updated.
* `PushServiceWebSocket` uses this to drop incoming updates with older
* versions.
*/
receivedPushMessage: function(keyID, message, cryptoParams, updateFunc) {
console.debug("receivedPushMessage()");
Services.telemetry.getHistogramById("PUSH_API_NOTIFICATION_RECEIVED").add();
let shouldNotify = false;
return this.getByKeyID(keyID).then(record => {
if (!record) {
this._recordDidNotNotify(kDROP_NOTIFICATION_REASON_KEY_NOT_FOUND);
throw new Error("No record for key ID " + keyID);
}
return record.getLastVisit();
}).then(lastVisit => {
// As a special case, don't notify the service worker if the user
// cleared their history.
shouldNotify = isFinite(lastVisit);
if (!shouldNotify) {
this._recordDidNotNotify(kDROP_NOTIFICATION_REASON_NO_HISTORY);
}
return this._db.update(keyID, record => {
let newRecord = updateFunc(record);
if (!newRecord) {
this._recordDidNotNotify(kDROP_NOTIFICATION_REASON_NO_VERSION_INCREMENT);
return null;
}
// Because `unregister` is advisory only, we can still receive messages
// for stale Simple Push registrations from the server. To work around
// this, we check if the record has expired before *and* after updating
// the quota.
if (newRecord.isExpired()) {
console.error("receivedPushMessage: Ignoring update for expired key ID",
keyID);
return null;
}
newRecord.receivedPush(lastVisit);
return newRecord;
});
}).then(record => {
var notified = false;
if (!record) {
return notified;
}
let decodedPromise;
if (cryptoParams) {
decodedPromise = PushCrypto.decodeMsg(
message,
record.p256dhPrivateKey,
cryptoParams.dh,
cryptoParams.salt,
cryptoParams.rs
);
} else {
decodedPromise = Promise.resolve(null);
}
return decodedPromise.then(message => {
if (shouldNotify) {
notified = this._notifyApp(record, message);
}
// Update quota after the delay, at which point
// we check for visible notifications.
setTimeout(() => this._updateQuota(keyID),
prefs.get("quotaUpdateDelay"));
return notified;
});
}).catch(error => {
console.error("receivedPushMessage: Error notifying app", error);
});
},
_updateQuota: function(keyID) {
console.debug("updateQuota()");
this._db.update(keyID, record => {
// Record may have expired from an earlier quota update.
if (record.isExpired()) {
console.debug(
"updateQuota: Trying to update quota for expired record", record);
return null;
}
// If there are visible notifications, don't apply the quota penalty
// for the message.
if (!this._visibleNotifications.has(record.uri.prePath)) {
record.reduceQuota();
}
return record;
}).then(record => {
if (record && record.isExpired()) {
this._recordDidNotNotify(kDROP_NOTIFICATION_REASON_EXPIRED);
// Drop the registration in the background. If the user returns to the
// site, the service worker will be notified on the next `idle-daily`
// event.
this._backgroundUnregister(record);
}
if (this._updateQuotaTestCallback) {
// Callback so that test may be notified when the quota update is complete.
this._updateQuotaTestCallback();
}
}).catch(error => {
console.debug("updateQuota: Error while trying to update quota", error);
});
},
_notificationForOriginShown(origin) {
console.debug("notificationForOriginShown()", origin);
let count;
if (this._visibleNotifications.has(origin)) {
count = this._visibleNotifications.get(origin);
} else {
count = 0;
}
this._visibleNotifications.set(origin, count + 1);
},
_notificationForOriginClosed(origin) {
console.debug("notificationForOriginClosed()", origin);
let count;
if (this._visibleNotifications.has(origin)) {
count = this._visibleNotifications.get(origin);
} else {
console.debug("notificationForOriginClosed: closing notification that has not been shown?");
return;
}
if (count > 1) {
this._visibleNotifications.set(origin, count - 1);
} else {
this._visibleNotifications.delete(origin);
}
},
_notifyApp: function(aPushRecord, message) {
if (!aPushRecord || !aPushRecord.scope ||
aPushRecord.originAttributes === undefined) {
console.error("notifyApp: Invalid record", aPushRecord);
return false;
}
console.debug("notifyApp()", aPushRecord.scope);
// Notify XPCOM observers.
let notification = Cc["@mozilla.org/push/ObserverNotification;1"]
.createInstance(Ci.nsIPushObserverNotification);
notification.pushEndpoint = aPushRecord.pushEndpoint;
notification.version = aPushRecord.version;
let payload = ArrayBuffer.isView(message) ?
new Uint8Array(message.buffer) : message;
if (payload) {
notification.data = "";
for (let i = 0; i < payload.length; i++) {
notification.data += String.fromCharCode(payload[i]);
}
}
notification.lastPush = aPushRecord.lastPush;
notification.pushCount = aPushRecord.pushCount;
Services.obs.notifyObservers(
notification,
"push-notification",
aPushRecord.scope
);
// If permission has been revoked, trash the message.
if (!aPushRecord.hasPermission()) {
console.warn("notifyApp: Missing push permission", aPushRecord);
return false;
}
let data = {
payload: payload,
originAttributes: aPushRecord.originAttributes,
scope: aPushRecord.scope
};
Services.telemetry.getHistogramById("PUSH_API_NOTIFY").add();
this._notifyListeners('push', data);
return true;
},
getByKeyID: function(aKeyID) {
return this._db.getByKeyID(aKeyID);
},
getAllUnexpired: function() {
return this._db.getAllUnexpired();
},
_sendRequest: function(action, aRecord) {
if (this._state == PUSH_SERVICE_CONNECTION_DISABLE) {
return Promise.reject(new Error("Push service disabled"));
} else if (this._state == PUSH_SERVICE_ACTIVE_OFFLINE) {
if (this._service.serviceType() == "WebSocket" && action == "unregister") {
return Promise.resolve();
}
return Promise.reject(new Error("Push service offline"));
}
return this._service.request(action, aRecord);
},
/**
* Called on message from the child process. aPageRecord is an object sent by
* navigator.push, identifying the sending page and other fields.
*/
_registerWithServer: function(aPageRecord) {
console.debug("registerWithServer()", aPageRecord);
Services.telemetry.getHistogramById("PUSH_API_SUBSCRIBE_ATTEMPT").add();
return this._sendRequest("register", aPageRecord)
.then(record => this._onRegisterSuccess(record),
err => this._onRegisterError(err))
.then(record => {
this._deletePendingRequest(aPageRecord);
return record.toSubscription();
}, err => {
this._deletePendingRequest(aPageRecord);
throw err;
});
},
_sendUnregister: function(aRecord) {
Services.telemetry.getHistogramById("PUSH_API_UNSUBSCRIBE_ATTEMPT").add();
return this._sendRequest("unregister", aRecord).then(function(v) {
Services.telemetry.getHistogramById("PUSH_API_UNSUBSCRIBE_SUCCEEDED").add();
return v;
}).catch(function(e) {
Services.telemetry.getHistogramById("PUSH_API_UNSUBSCRIBE_FAILED").add();
return Promise.reject(e);
});
},
/**
* Exceptions thrown in _onRegisterSuccess are caught by the promise obtained
* from _service.request, causing the promise to be rejected instead.
*/
_onRegisterSuccess: function(aRecord) {
console.debug("_onRegisterSuccess()");
return this._db.put(aRecord)
.then(record => {
Services.telemetry.getHistogramById("PUSH_API_SUBSCRIBE_SUCCEEDED").add();
return record;
})
.catch(error => {
Services.telemetry.getHistogramById("PUSH_API_SUBSCRIBE_FAILED").add()
// Unable to save. Destroy the subscription in the background.
this._backgroundUnregister(aRecord);
throw error;
});
},
/**
* Exceptions thrown in _onRegisterError are caught by the promise obtained
* from _service.request, causing the promise to be rejected instead.
*/
_onRegisterError: function(reply) {
console.debug("_onRegisterError()");
Services.telemetry.getHistogramById("PUSH_API_SUBSCRIBE_FAILED").add()
if (!reply.error) {
console.warn("onRegisterError: Called without valid error message!",
reply.error);
throw new Error("Registration error");
}
throw reply.error;
},
receiveMessage: function(aMessage) {
console.debug("receiveMessage()", aMessage.name);
if (kCHILD_PROCESS_MESSAGES.indexOf(aMessage.name) == -1) {
console.debug("receiveMessage: Invalid message from child",
aMessage.name);
return;
}
if (aMessage.name === "Push:RegisterEventNotificationListener") {
console.debug("receiveMessage: Adding child listener");
this._childListeners.push(aMessage.target);
return;
}
if (aMessage.name === "child-process-shutdown") {
console.debug("receiveMessage: Possibly removing child listener");
for (var i = this._childListeners.length - 1; i >= 0; --i) {
if (this._childListeners[i] == aMessage.target) {
console.debug("receiveMessage: Removed child listener");
this._childListeners.splice(i, 1);
}
}
console.debug("receiveMessage: Clearing notifications from child");
this._visibleNotifications.clear();
return;
}
if (aMessage.name === "Push:NotificationForOriginShown") {
console.debug("receiveMessage: Notification shown from child");
this._notificationForOriginShown(aMessage.data);
return;
}
if (aMessage.name === "Push:NotificationForOriginClosed") {
console.debug("receiveMessage: Notification closed from child");
this._notificationForOriginClosed(aMessage.data);
return;
}
if (!aMessage.target.assertPermission("push")) {
console.debug("receiveMessage: Got message from a child process that",
"does not have 'push' permission");
return null;
}
let mm = aMessage.target.QueryInterface(Ci.nsIMessageSender);
let name = aMessage.name.slice("Push:".length);
Promise.resolve().then(_ => {
let pageRecord = this._validatePageRecord(aMessage);
return this[name.toLowerCase()](pageRecord);
}).then(result => {
mm.sendAsyncMessage("PushService:" + name + ":OK", {
requestID: aMessage.data.requestID,
result: result,
});
}, error => {
console.error("receiveMessage: Error handling message", aMessage, error);
mm.sendAsyncMessage("PushService:" + name + ":KO", {
requestID: aMessage.data.requestID,
});
}).catch(error => {
console.error("receiveMessage: Error sending reply", error);
});
},
_validatePageRecord: function(aMessage) {
let principal = aMessage.principal;
if (!principal) {
throw new Error("Missing message principal");
}
let pageRecord = aMessage.data;
if (!pageRecord.scope) {
throw new Error("Missing page record scope");
}
pageRecord.originAttributes =
ChromeUtils.originAttributesToSuffix(principal.originAttributes);
return pageRecord;
},
register: function(aPageRecord) {
console.debug("register()", aPageRecord);
if (!aPageRecord.scope || aPageRecord.originAttributes === undefined) {
return Promise.reject(new Error("Invalid page record"));
}
return this._checkActivated()
.then(_ => this._db.getByIdentifiers(aPageRecord))
.then(record => {
if (!record) {
return this._lookupOrPutPendingRequest(aPageRecord);
}
if (record.isExpired()) {
return record.quotaChanged().then(isChanged => {
if (isChanged) {
// If the user revisited the site, drop the expired push
// registration and re-register.
return this.dropRegistrationAndNotifyApp(record.keyID);
}
throw new Error("Push subscription expired");
}).then(_ => this._lookupOrPutPendingRequest(aPageRecord));
}
return record.toSubscription();
});
},
/**
* Called on message from the child process.
*
* Why is the record being deleted from the local database before the server
* is told?
*
* Unregistration is for the benefit of the app and the AppServer
* so that the AppServer does not keep pinging a channel the UserAgent isn't
* watching The important part of the transaction in this case is left to the
* app, to tell its server of the unregistration. Even if the request to the
* PushServer were to fail, it would not affect correctness of the protocol,
* and the server GC would just clean up the channelID/subscription
* eventually. Since the appserver doesn't ping it, no data is lost.
*
* If rather we were to unregister at the server and update the database only
* on success: If the server receives the unregister, and deletes the
* channelID/subscription, but the response is lost because of network
* failure, the application is never informed. In addition the application may
* retry the unregister when it fails due to timeout (websocket) or any other
* reason at which point the server will say it does not know of this
* unregistration. We'll have to make the registration/unregistration phases
* have retries and attempts to resend messages from the server, and have the
* client acknowledge. On a server, data is cheap, reliable notification is
* not.
*/
unregister: function(aPageRecord) {
console.debug("unregister()", aPageRecord);
if (!aPageRecord.scope || aPageRecord.originAttributes === undefined) {
return Promise.reject(new Error("Invalid page record"));
}
return this._checkActivated()
.then(_ => this._db.getByIdentifiers(aPageRecord))
.then(record => {
if (record === undefined) {
return false;
}
return Promise.all([
this._sendUnregister(record),
this._db.delete(record.keyID),
]).then(() => true);
});
},
_clearAll: function _clearAll() {
return this._checkActivated()
.then(_ => this._db.drop())
.catch(_ => Promise.resolve());
},
_clearForDomain: function(domain) {
/**
* Copied from ForgetAboutSite.jsm.
*
* Returns true if the string passed in is part of the root domain of the
* current string. For example, if this is "www.mozilla.org", and we pass in
* "mozilla.org", this will return true. It would return false the other way
* around.
*/
function hasRootDomain(str, aDomain)
{
let index = str.indexOf(aDomain);
// If aDomain is not found, we know we do not have it as a root domain.
if (index == -1)
return false;
// If the strings are the same, we obviously have a match.
if (str == aDomain)
return true;
// Otherwise, we have aDomain as our root domain iff the index of aDomain is
// aDomain.length subtracted from our length and (since we do not have an
// exact match) the character before the index is a dot or slash.
let prevChar = str[index - 1];
return (index == (str.length - aDomain.length)) &&
(prevChar == "." || prevChar == "/");
}
let clear = (db, domain) => {
db.clearIf(record => {
return hasRootDomain(record.uri.prePath, domain);
});
}
return this._checkActivated()
.then(_ => clear(this._db, domain))
.catch(e => {
console.warn("clearForDomain: Error forgetting about domain", e);
return Promise.resolve();
});
},
registration: function(aPageRecord) {
console.debug("registration()");
if (!aPageRecord.scope || aPageRecord.originAttributes === undefined) {
return Promise.reject(new Error("Invalid page record"));
}
return this._checkActivated()
.then(_ => this._db.getByIdentifiers(aPageRecord))
.then(record => {
if (!record) {
return null;
}
if (record.isExpired()) {
return record.quotaChanged().then(isChanged => {
if (isChanged) {
return this.dropRegistrationAndNotifyApp(record.keyID).then(_ => null);
}
return null;
});
}
return record.toSubscription();
});
},
_dropExpiredRegistrations: function() {
console.debug("dropExpiredRegistrations()");
return this._db.getAllExpired().then(records => {
return Promise.all(records.map(record =>
record.quotaChanged().then(isChanged => {
if (isChanged) {
// If the user revisited the site, drop the expired push
// registration and notify the associated service worker.
return this.dropRegistrationAndNotifyApp(record.keyID);
}
}).catch(error => {
console.error("dropExpiredRegistrations: Error dropping registration",
record.keyID, error);
})
));
});
},
_onPermissionChange: function(subject, data) {
console.debug("onPermissionChange()");
if (data == "cleared") {
// If the permission list was cleared, drop all registrations
// that are subject to quota.
return this._db.clearIf(record => {
if (record.quotaApplies()) {
if (!record.isExpired()) {
// Drop the registration in the background.
this._backgroundUnregister(record);
}
return true;
}
return false;
});
}
let permission = subject.QueryInterface(Ci.nsIPermission);
if (permission.type != "desktop-notification") {
return Promise.resolve();
}
return this._updatePermission(permission, data);
},
_updatePermission: function(permission, type) {
console.debug("updatePermission()");
let isAllow = permission.capability ==
Ci.nsIPermissionManager.ALLOW_ACTION;
let isChange = type == "added" || type == "changed";
if (isAllow && isChange) {
// Permission set to "allow". Drop all expired registrations for this
// site, notify the associated service workers, and reset the quota
// for active registrations.
return this._reduceByPrincipal(
permission.principal,
(subscriptionChanges, record, cursor) => {
this._permissionAllowed(subscriptionChanges, record, cursor);
return subscriptionChanges;
},
[]
).then(subscriptionChanges => {
this.notifySubscriptionChanges(subscriptionChanges);
});
} else if (isChange || (isAllow && type == "deleted")) {
// Permission set to "block" or "always ask," or "allow" permission
// removed. Expire all registrations for this site.
return this._reduceByPrincipal(
permission.principal,
(memo, record, cursor) => this._permissionDenied(record, cursor)
);
}
return Promise.resolve();
},
_reduceByPrincipal: function(principal, callback, initialValue) {
return this._db.reduceByOrigin(
principal.URI.prePath,
ChromeUtils.originAttributesToSuffix(principal.originAttributes),
callback,
initialValue
);
},
/**
* The update function called for each registration record if the push
* permission is revoked. We only expire the record so we can notify the
* service worker as soon as the permission is reinstated. If we just
* deleted the record, the worker wouldn't be notified until the next visit
* to the site.
*
* @param {PushRecord} record The record to expire.
* @param {IDBCursor} cursor The IndexedDB cursor.
*/
_permissionDenied: function(record, cursor) {
console.debug("permissionDenied()");
if (!record.quotaApplies() || record.isExpired()) {
// Ignore already-expired records.
return;
}
// Drop the registration in the background.
this._backgroundUnregister(record);
record.setQuota(0);
cursor.update(record);
},
/**
* The update function called for each registration record if the push
* permission is granted. If the record has expired, it will be dropped;
* otherwise, its quota will be reset to the default value.
*
* @param {Array} subscriptionChanges A list of records whose associated
* service workers should be notified once the transaction has committed.
* @param {PushRecord} record The record to update.
* @param {IDBCursor} cursor The IndexedDB cursor.
*/
_permissionAllowed: function(subscriptionChanges, record, cursor) {
console.debug("permissionAllowed()");
if (!record.quotaApplies()) {
return;
}
if (record.isExpired()) {
// If the registration has expired, drop and notify the worker
// unconditionally.
subscriptionChanges.push(record);
cursor.delete();
return;
}
record.resetQuota();
cursor.update(record);
},
};