Magic-Mirror/gecko-dev
1
0
Fork 0
mirror of https://github.com/mozilla/gecko-dev.git synced 2025-02-17 14:25:49 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Bug 1150812 - Add Http2 Push service. r=nsm, r=mt

Browse Source
This commit is contained in:
Dragana Damjanovic 2015-06-03 08:06:00 -04:00
parent 2ce8502cf6
commit af2c1ad891
20 changed files with 960 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
dom/push/PushDB.jsm
View File

@ -45,20 +45,19 @@ this.PushDB.prototype = {
},
/*
* @param aChannelRecord
* @param aRecord
* The record to be added.
*/
put: function(aChannelRecord) {
debug("put()" + JSON.stringify(aChannelRecord));
put: function(aRecord) {
debug("put()" + JSON.stringify(aRecord));
return new Promise((resolve, reject) =>
this.newTxn(
"readwrite",
this._dbStoreName,
function txnCb(aTxn, aStore) {
debug("Going to put " + aChannelRecord.channelID);
aStore.put(aChannelRecord).onsuccess = function setTxnResult(aEvent) {
aStore.put(aRecord).onsuccess = function setTxnResult(aEvent) {
debug("Request successful. Updated record ID: " +
aEvent.target.result);
};
@ -70,10 +69,10 @@ this.PushDB.prototype = {
},
/*
* @param aChannelID
* @param aKeyID
* The ID of record to be deleted.
*/
delete: function(aChannelID) {
delete: function(aKeyID) {
debug("delete()");
return new Promise((resolve, reject) =>
@ -81,8 +80,8 @@ this.PushDB.prototype = {
"readwrite",
this._dbStoreName,
function txnCb(aTxn, aStore) {
debug("Going to delete " + aChannelID);
aStore.delete(aChannelID);
debug("Going to delete " + aKeyID);
aStore.delete(aKeyID);
},
resolve,
reject
@ -127,8 +126,8 @@ this.PushDB.prototype = {
);
},
getByChannelID: function(aChannelID) {
debug("getByChannelID()");
getByKeyID: function(aKeyID) {
debug("getByKeyID()");
return new Promise((resolve, reject) =>
this.newTxn(
@ -137,7 +136,7 @@ this.PushDB.prototype = {
function txnCb(aTxn, aStore) {
aTxn.result = undefined;
aStore.get(aChannelID).onsuccess = function setTxnResult(aEvent) {
aStore.get(aKeyID).onsuccess = function setTxnResult(aEvent) {
aTxn.result = aEvent.target.result;
debug("Fetch successful " + aEvent.target.result);
};
@ -171,8 +170,8 @@ this.PushDB.prototype = {
);
},
getAllChannelIDs: function() {
debug("getAllChannelIDs()");
getAllKeyIDs: function() {
debug("getAllKeyIDs()");
return new Promise((resolve, reject) =>
this.newTxn(

119
dom/push/PushService.jsm
View File

@ -27,9 +27,10 @@ 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");
// Currently supported protocols: WebSocket.
const CONNECTION_PROTOCOLS = [PushServiceWebSocket];
const CONNECTION_PROTOCOLS = [PushServiceWebSocket, PushServiceHttp2];
XPCOMUtils.defineLazyModuleGetter(this, "AlarmService",
"resource://gre/modules/AlarmService.jsm");
@ -160,11 +161,11 @@ this.PushService = {
// Disconnect first.
this._service.disconnect();
}
this._db.getAllChannelIDs()
.then(channelIDs => {
if (channelIDs.length > 0) {
this._db.getAllKeyIDs()
.then(keyIDs => {
if (keyIDs.length > 0) {
// if there are request waiting
this._service.connect(channelIDs);
this._service.connect(keyIDs);
}
});
this._setState(PUSH_SERVICE_RUNNING);
@ -245,7 +246,7 @@ this.PushService = {
this._db.getByScope(scope)
.then(record =>
Promise.all([
this._db.delete(record.channelID),
this._db.delete(this._service.getKeyFromRecord(record)),
this._sendRequest("unregister", record)
])
).catch(_ => {
@ -500,8 +501,11 @@ this.PushService = {
msgName => ppmm.removeMessageListener(msgName, this)
);
}
this._service.disconnect();
this._service.uninit();
this._service = null;
this.stopAlarm();
if (!this._db) {
return Promise.resolve();
@ -576,7 +580,7 @@ this.PushService = {
ignoreTimezone: true
},
() => {
if (this._service) {
if (this._state > PUSH_SERVICE_ACTIVATING) {
this._service.onAlarmFired();
}
}, (alarmID) => {
@ -610,7 +614,7 @@ this.PushService = {
_notifyAllAppsRegister: function() {
debug("notifyAllAppsRegister()");
// records are objects describing the registration as stored in IndexedDB.
return this._db.getAllChannelIDs()
return this._db.getAllKeyIDs()
.then(records => {
let scopes = new Set();
for (let record of records) {
@ -636,8 +640,50 @@ this.PushService = {
});
},
dropRegistrationAndNotifyApp: function(aKeyId) {
return this._db.getByKeyID(aKeyId)
.then(record => {
let globalMM = Cc['@mozilla.org/globalmessagemanager;1']
.getService(Ci.nsIMessageListenerManager);
Services.obs.notifyObservers(
null,
"push-subscription-change",
record.scope
);
let data = {
originAttributes: {}, // TODO bug 1166350
scope: record.scope
};
globalMM.broadcastAsyncMessage('pushsubscriptionchange', data);
})
.then(_ => this._db.delete(aKeyId));
},
updateRegistrationAndNotifyApp: function(aOldKey, aRecord) {
return this._db.delete(aOldKey)
.then(_ => this._db.put(aRecord)
.then(record => {
let globalMM = Cc['@mozilla.org/globalmessagemanager;1']
.getService(Ci.nsIMessageListenerManager);
Services.obs.notifyObservers(
null,
"push-subscription-change",
record.scope
);
let data = {
originAttributes: {}, // TODO bug 1166350
scope: record.scope
};
globalMM.broadcastAsyncMessage('pushsubscriptionchange', data);
}));
},
receivedPushMessage: function(aPushRecord, message) {
this._updatePushRecord(aPushRecord)
this._db.put(aPushRecord)
.then(_ => this._notifyApp(aPushRecord, message));
},
@ -684,17 +730,12 @@ this.PushService = {
globalMM.broadcastAsyncMessage('push', data);
},
_updatePushRecord: function(aPushRecord) {
debug("updatePushRecord()");
return this._db.put(aPushRecord);
getByKeyID: function(aKeyID) {
return this._db.getByKeyID(aKeyID);
},
getByChannelID: function(aChannelID) {
return this._db.getByChannelID(aChannelID);
},
getAllChannelIDs: function() {
return this._db.getAllChannelIDs();
getAllKeyIDs: function() {
return this._db.getAllKeyIDs();
},
_sendRequest(action, aRecord) {
@ -753,7 +794,7 @@ this.PushService = {
_onRegisterSuccess: function(aRecord) {
debug("_onRegisterSuccess()");
return this._updatePushRecord(aRecord)
return this._db.put(aRecord)
.then(_ => aRecord, error => {
// Unable to save.
this._sendRequest("unregister", aRecord);
@ -793,10 +834,8 @@ this.PushService = {
this._register(aPageRecord)
.then(pushRecord => {
let message = {
requestID: aPageRecord.requestID,
pushEndpoint: pushRecord.pushEndpoint
};
let message = this._service.prepareRegister(pushRecord);
message.requestID = aPageRecord.requestID;
aMessageManager.sendAsyncMessage("PushService:Register:OK", message);
}, error => {
let message = {
@ -818,18 +857,19 @@ this.PushService = {
* 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 eventually. Since the
* appserver doesn't ping it, no data is lost.
* 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, 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 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.
* 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) {
debug("unregisterWithServer()");
@ -846,10 +886,10 @@ this.PushService = {
throw "NotFoundError";
}
// Let's be nice to the server and try to inform it, but we don't care
// about the reply.
this._sendRequest("unregister", record);
this._db.delete(record.channelID);
return Promise.all([
this._sendRequest("unregister", record),
this._db.delete(this._service.getKeyFromRecord(record))
]);
});
},
@ -892,12 +932,7 @@ this.PushService = {
if (!pushRecord) {
return null;
}
return {
pushEndpoint: pushRecord.pushEndpoint,
version: pushRecord.version,
lastPush: pushRecord.lastPush,
pushCount: pushRecord.pushCount
};
return this._service.prepareRegistration(pushRecord);
});
},

820
dom/push/PushServiceHttp2.jsm Normal file
View File

@ -0,0 +1,820 @@
/* 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/IndexedDBHelper.jsm");
Cu.import("resource://gre/modules/Timer.jsm");
Cu.import("resource://gre/modules/Preferences.jsm");
Cu.import("resource://gre/modules/Promise.jsm");
this.EXPORTED_SYMBOLS = ["PushServiceHttp2"];
const prefs = new Preferences("dom.push.");
// Don't modify this, instead set dom.push.debug.
// Set debug first so that all debugging actually works.
var gDebuggingEnabled = prefs.get("debug");
function debug(s) {
if (gDebuggingEnabled) {
dump("-*- PushServiceHttp2.jsm: " + s + "\n");
}
}
const kPUSHHTTP2DB_DB_NAME = "pushHttp2";
const kPUSHHTTP2DB_DB_VERSION = 1; // Change this if the IndexedDB format changes
const kPUSHHTTP2DB_STORE_NAME = "pushHttp2";
/**
* A proxy between the PushService and connections listening for incoming push
* messages. The PushService can silence messages from the connections by
* setting PushSubscriptionListener._pushService to null. This is required
* because it can happen that there is an outstanding push message that will
* be send on OnStopRequest but the PushService may not be interested in these.
* It's easier to stop listening than to have checks at specific points.
*/
var PushSubscriptionListener = function(pushService, uri) {
debug("Creating a new pushSubscription listener.");
this._pushService = pushService;
this.uri = uri;
};
PushSubscriptionListener.prototype = {
QueryInterface: function (aIID) {
if (aIID.equals(Ci.nsIHttpPushListener) ||
aIID.equals(Ci.nsIStreamListener)) {
return this;
}
throw Components.results.NS_ERROR_NO_INTERFACE;
},
getInterface: function(aIID) {
return this.QueryInterface(aIID);
},
onStartRequest: function(aRequest, aContext) {
debug("PushSubscriptionListener onStartRequest()");
// We do not do anything here.
},
onDataAvailable: function(aRequest, aContext, aStream, aOffset, aCount) {
debug("PushSubscriptionListener onDataAvailable()");
// Nobody should send data, but just to be sure, otherwise necko will
// complain.
if (aCount === 0) {
return;
}
let inputStream = Cc["@mozilla.org/scriptableinputstream;1"]
.createInstance(Ci.nsIScriptableInputStream);
inputStream.init(aStream);
var data = inputStream.read(aCount);
},
onStopRequest: function(aRequest, aContext, aStatusCode) {
debug("PushSubscriptionListener onStopRequest()");
if (!this._pushService) {
return;
}
this._pushService.connOnStop(aRequest,
Components.isSuccessCode(aStatusCode),
this.uri);
},
onPush: function(associatedChannel, pushChannel) {
debug("PushSubscriptionListener onPush()");
var pushChannelListener = new PushChannelListener(this);
pushChannel.asyncOpen(pushChannelListener, pushChannel);
},
disconnect: function() {
this._pushService = null;
}
};
/**
* The listener for pushed messages. The message data is collected in
* OnDataAvailable and send to the app in OnStopRequest.
*/
var PushChannelListener = function(pushSubscriptionListener) {
debug("Creating a new push channel listener.");
this._mainListener = pushSubscriptionListener;
};
PushChannelListener.prototype = {
_message: null,
_ackUri: null,
onStartRequest: function(aRequest, aContext) {
this._ackUri = aRequest.URI.spec;
},
onDataAvailable: function(aRequest, aContext, aStream, aOffset, aCount) {
debug("push channel listener onDataAvailable()");
if (aCount === 0) {
return;
}
let inputStream = Cc["@mozilla.org/scriptableinputstream;1"]
.createInstance(Ci.nsIScriptableInputStream);
inputStream.init(aStream);
if (!this._message) {
this._message = inputStream.read(aCount);
} else {
this._message.concat(inputStream.read(aCount));
}
},
onStopRequest: function(aRequest, aContext, aStatusCode) {
debug("push channel listener onStopRequest() status code:" + aStatusCode);
if (Components.isSuccessCode(aStatusCode) &&
this._mainListener &&
this._mainListener._pushService) {
this._mainListener._pushService._pushChannelOnStop(this._mainListener.uri,
this._ackUri,
this._message);
}
}
};
var PushServiceDelete = function(resolve, reject) {
this._resolve = resolve;
this._reject = reject;
};
PushServiceDelete.prototype = {
onStartRequest: function(aRequest, aContext) {},
onDataAvailable: function(aRequest, aContext, aStream, aOffset, aCount) {
// Nobody should send data, but just to be sure, otherwise necko will
// complain.
if (aCount === 0) {
return;
}
let inputStream = Cc["@mozilla.org/scriptableinputstream;1"]
.createInstance(Ci.nsIScriptableInputStream);
inputStream.init(aStream);
var data = inputStream.read(aCount);
},
onStopRequest: function(aRequest, aContext, aStatusCode) {
if (Components.isSuccessCode(aStatusCode)) {
this._resolve();
} else {
this._reject({status: 0, error: "NetworkError"});
}
}
};
var SubscriptionListener = function(aSubInfo, aServerURI, aPushServiceHttp2) {
debug("Creating a new subscription listener.");
this._subInfo = aSubInfo;
this._data = '';
this._serverURI = aServerURI;
this._service = aPushServiceHttp2;
};
SubscriptionListener.prototype = {
onStartRequest: function(aRequest, aContext) {},
onDataAvailable: function(aRequest, aContext, aStream, aOffset, aCount) {
debug("subscription listener onDataAvailable()");
// We do not expect any data, but necko will complain if we do not consume
// it.
if (aCount === 0) {
return;
}
let inputStream = Cc["@mozilla.org/scriptableinputstream;1"]
.createInstance(Ci.nsIScriptableInputStream);
inputStream.init(aStream);
this._data.concat(inputStream.read(aCount));
},
onStopRequest: function(aRequest, aContext, aStatus) {
debug("subscription listener onStopRequest()");
// Check if pushService is still active.
if (!this._service.hasmainPushService()) {
this._subInfo.reject({error: "Service deactivated"});
return;
}
if (!Components.isSuccessCode(aStatus)) {
this._subInfo.reject({error: "Error status" + aStatus});
return;
}
var statusCode = aRequest.QueryInterface(Ci.nsIHttpChannel).responseStatus;
if (Math.floor(statusCode / 100) == 5) {
if (this._subInfo.retries < prefs.get("http2.maxRetries")) {
this._subInfo.retries++;
var retryAfter = retryAfterParser(aRequest);
setTimeout(this._service.retrySubscription.bind(this._service,
this._subInfo),
retryAfter);
} else {
this._subInfo.reject({error: "Error response code: " + statusCode });
}
return;
} else if (statusCode != 201) {
this._subInfo.reject({error: "Error response code: " + statusCode });
return;
}
var subscriptionUri;
try {
subscriptionUri = aRequest.getResponseHeader("location");
} catch (err) {
this._subInfo.reject({error: "Return code 201, but the answer is bogus"});
return;
}
debug("subscriptionUri: " + subscriptionUri);
var linkList;
try {
linkList = aRequest.getResponseHeader("link");
} catch (err) {
this._subInfo.reject({error: "Return code 201, but the answer is bogus"});
return;
}
var linkParserResult = linkParser(linkList, this._serverURI);
if (linkParserResult.error) {
this._subInfo.reject(linkParserResult);
return;
}
if (!subscriptionUri) {
this._subInfo.reject({error: "Return code 201, but the answer is bogus," +
" missing subscriptionUri"});
return;
}
try {
let uriTry = Services.io.newURI(subscriptionUri, null, null);
} catch (e) {
debug("Invalid URI " + subscriptionUri);
this._subInfo.reject({error: "Return code 201, but URI is bogus. " +
subscriptionUri});
return;
}
var reply = {
subscriptionUri: subscriptionUri,
pushEndpoint: linkParserResult.pushEndpoint,
pushReceiptEndpoint: linkParserResult.pushReceiptEndpoint,
pageURL: this._subInfo.record.pageURL,
scope: this._subInfo.record.scope,
pushCount: 0,
lastPush: 0
};
this._subInfo.resolve(reply);
},
};
function retryAfterParser(aRequest) {
var retryAfter = 0;
try {
var retryField = aRequest.getResponseHeader("retry-after");
if (isNaN(retryField)) {
retryAfter = Date.parse(retryField) - (new Date().getTime());
} else {
retryAfter = parseInt(retryField, 10) * 1000;
}
retryAfter = (retryAfter > 0) ? retryAfter : 0;
} catch(e) {}
return retryAfter;
}
function linkParser(linkHeader, serverURI) {
var linkList = linkHeader.split(',');
if ((linkList.length < 1)) {
return {error: "Return code 201, but the answer is bogus"};
}
var pushEndpoint;
var pushReceiptEndpoint;
linkList.forEach(link => {
var linkElems = link.split(';');
if (linkElems.length == 2) {
if (linkElems[1].trim() === 'rel="urn:ietf:params:push"') {
pushEndpoint = linkElems[0].substring(linkElems[0].indexOf('<') + 1,
linkElems[0].indexOf('>'));
} else if (linkElems[1].trim() === 'rel="urn:ietf:params:push:receipt"') {
pushReceiptEndpoint = linkElems[0].substring(linkElems[0].indexOf('<') + 1,
linkElems[0].indexOf('>'));
}
}
});
debug("pushEndpoint: " + pushEndpoint);
debug("pushReceiptEndpoint: " + pushReceiptEndpoint);
// Missing pushReceiptEndpoint is allowed.
if (!pushEndpoint) {
return {error: "Return code 201, but the answer is bogus, missing" +
" pushEndpoint"};
}
var uri;
var resUri = [];
try {
[pushEndpoint, pushReceiptEndpoint].forEach(u => {
if (u) {
uri = u;
resUri[u] = Services.io.newURI(uri, null, serverURI);
}
});
} catch (e) {
debug("Invalid URI " + uri);
return {error: "Return code 201, but URI is bogus. " + uri};
}
return {
pushEndpoint: resUri[pushEndpoint].spec,
pushReceiptEndpoint: (pushReceiptEndpoint) ? resUri[pushReceiptEndpoint].spec
: ""
};
}
/**
* The implementation of the WebPush.
*/
this.PushServiceHttp2 = {
_mainPushService: null,
_serverURI: null,
// Keep information about all connections, e.g. the channel, listener...
_conns: {},
_started: false,
upgradeSchema: function(aTransaction,
aDb,
aOldVersion,
aNewVersion,
aDbInstance) {
debug("upgradeSchemaHttp2()");
let objectStore = aDb.createObjectStore(aDbInstance._dbStoreName,
{ keyPath: "subscriptionUri" });
// index to fetch records based on endpoints. used by unregister
objectStore.createIndex("pushEndpoint", "pushEndpoint", { unique: true });
// index to fetch records per scope, so we can identify endpoints
// associated with an app.
objectStore.createIndex("scope", "scope", { unique: true });
},
getKeyFromRecord: function(aRecord) {
return aRecord.subscriptionUri;
},
newPushDB: function() {
return new PushDB(kPUSHHTTP2DB_DB_NAME,
kPUSHHTTP2DB_DB_VERSION,
kPUSHHTTP2DB_STORE_NAME,
this.upgradeSchema);
},
hasmainPushService: function() {
return this._mainPushService !== null;
},
checkServerURI: function(serverURL) {
if (!serverURL) {
debug("No dom.push.serverURL found!");
return;
}
let uri;
try {
uri = Services.io.newURI(serverURL, null, null);
} catch(e) {
debug("Error creating valid URI from dom.push.serverURL (" +
serverURL + ")");
return null;
}
if (uri.scheme !== "https") {
debug("Unsupported websocket scheme " + uri.scheme);
return null;
}
return uri;
},
observe: function(aSubject, aTopic, aData) {
if (aTopic == "nsPref:changed") {
if (aData == "dom.push.debug") {
gDebuggingEnabled = prefs.get("debug");
}
}
},
connect: function(subscriptions) {
this.startConnections(subscriptions);
},
disconnect: function() {
this._shutdownConnections(false);
},
_makeChannel: function(aUri) {
var ios = Cc["@mozilla.org/network/io-service;1"]
.getService(Ci.nsIIOService);
var chan = ios.newChannel2(aUri,
null,
null,
null, // aLoadingNode
Services.scriptSecurityManager.getSystemPrincipal(),
null, // aTriggeringPrincipal
Ci.nsILoadInfo.SEC_NORMAL,
Ci.nsIContentPolicy.TYPE_OTHER)
.QueryInterface(Ci.nsIHttpChannel);
var loadGroup = Cc["@mozilla.org/network/load-group;1"]
.createInstance(Ci.nsILoadGroup);
chan.loadGroup = loadGroup;
return chan;
},
/**
* Subscribe new resource.
*/
_subscribeResource: function(aRecord) {
debug("subscribeResource()");
return new Promise((resolve, reject) => {
this._subscribeResourceInternal({record: aRecord,
resolve,
reject,
retries: 0});
})
.then(result => {
this._conns[result.subscriptionUri] = {channel: null,
listener: null,
countUnableToConnect: 0,
lastStartListening: 0,
waitingForAlarm: false};
this._listenForMsgs(result.subscriptionUri);
return result;
});
},
_subscribeResourceInternal: function(aSubInfo) {
debug("subscribeResource()");
var listener = new SubscriptionListener(aSubInfo,
this._serverURI,
this);
var chan = this._makeChannel(this._serverURI.spec);
chan.requestMethod = "POST";
try{
chan.asyncOpen(listener, null);
} catch(e) {
aSubInfo.reject({status: 0, error: "NetworkError"});
}
},
retrySubscription: function(aSubInfo) {
this._subscribeResourceInternal(aSubInfo);
},
_deleteResource: function(aUri) {
return new Promise((resolve,reject) => {
var chan = this._makeChannel(aUri);
chan.requestMethod = "DELETE";
try {
chan.asyncOpen(new PushServiceDelete(resolve, reject), null);
} catch(err) {
reject({status: 0, error: "NetworkError"});
}
});
},
/**
* Unsubscribe the resource with a subscription uri aSubscriptionUri.
* We can't do anything about it if it fails, so we don't listen for response.
*/
_unsubscribeResource: function(aSubscriptionUri) {
debug("unsubscribeResource()");
return this._deleteResource(aSubscriptionUri);
},
/**
* Start listening for messages.
*/
_listenForMsgs: function(aSubscriptionUri) {
debug("listenForMsgs() " + aSubscriptionUri);
if (!this._conns[aSubscriptionUri]) {
debug("We do not have this subscription " + aSubscriptionUri);
return;
}
var chan = this._makeChannel(aSubscriptionUri);
var conn = {};
conn.channel = chan;
var listener = new PushSubscriptionListener(this, aSubscriptionUri);
conn.listener = listener;
chan.notificationCallbacks = listener;
try {
chan.asyncOpen(listener, chan);
} catch (e) {
debug("Error connecting to push server. asyncOpen failed!");
conn.listener.disconnect();
chan.cancel(Cr.NS_ERROR_ABORT);
this._retryAfterBackoff(aSubscriptionUri, -1);
return;
}
this._conns[aSubscriptionUri].lastStartListening = Date.now();
this._conns[aSubscriptionUri].channel = conn.channel;
this._conns[aSubscriptionUri].listener = conn.listener;
},
_ackMsgRecv: function(aAckUri) {
debug("ackMsgRecv() " + aAckUri);
// We can't do anything about it if it fails,
// so we don't listen for response.
this._deleteResource(aAckUri);
},
init: function(aOptions, aMainPushService, aServerURL) {
debug("init()");
this._mainPushService = aMainPushService;
this._serverURI = aServerURL;
gDebuggingEnabled = prefs.get("debug");
prefs.observe("debug", this);
},
_retryAfterBackoff: function(aSubscriptionUri, retryAfter) {
debug("retryAfterBackoff()");
var resetRetryCount = prefs.get("http2.reset_retry_count_after_ms");
// If it was running for some time, reset retry counter.
if ((Date.now() - this._conns[aSubscriptionUri].lastStartListening) >
resetRetryCount) {
this._conns[aSubscriptionUri].countUnableToConnect = 0;
}
let maxRetries = prefs.get("http2.maxRetries");
if (this._conns[aSubscriptionUri].countUnableToConnect >= maxRetries) {
this._shutdownSubscription(aSubscriptionUri);
this._resubscribe(aSubscriptionUri);
return;
}
if (retryAfter !== -1) {
// This is a 5xx response.
// To respect RetryAfter header, setTimeout is used. setAlarm sets a
// cumulative alarm so it will not always respect RetryAfter header.
this._conns[aSubscriptionUri].countUnableToConnect++;
setTimeout(_ => this._listenForMsgs(aSubscriptionUri), retryAfter);
return;
}
// we set just one alarm because most probably all connection will go over
// a single TCP connection.
retryAfter = prefs.get("http2.retryInterval") *
Math.pow(2, this._conns[aSubscriptionUri].countUnableToConnect);
retryAfter = retryAfter * (0.8 + Math.random() * 0.4); // add +/-20%.
this._conns[aSubscriptionUri].countUnableToConnect++;
if (retryAfter === 0) {
setTimeout(_ => this._listenForMsgs(aSubscriptionUri), 0);
} else {
this._conns[aSubscriptionUri].waitingForAlarm = true;
this._mainPushService.setAlarm(retryAfter);
}
debug("Retry in " + retryAfter);
},
// Close connections.
_shutdownConnections: function(deleteInfo) {
debug("shutdownConnections()");
for (let subscriptionUri in this._conns) {
if (this._conns[subscriptionUri]) {
if (this._conns[subscriptionUri].listener) {
this._conns[subscriptionUri].listener._pushService = null;
}
if (this._conns[subscriptionUri].channel) {
try {
this._conns[subscriptionUri].channel.cancel(Cr.NS_ERROR_ABORT);
} catch (e) {}
}
this._conns[subscriptionUri].listener = null;
this._conns[subscriptionUri].channel = null;
this._conns[subscriptionUri].waitingForAlarm = false;
if (deleteInfo) {
delete this._conns[subscriptionUri];
}
}
}
},
// Start listening if subscriptions present.
startConnections: function(aSubscriptions) {
debug("startConnections() " + aSubscriptions.length);
for (let i = 0; i < aSubscriptions.length; i++) {
let record = aSubscriptions[i];
if (typeof this._conns[record.subscriptionUri] != "object") {
this._conns[record.subscriptionUri] = {channel: null,
listener: null,
countUnableToConnect: 0,
waitingForAlarm: false};
}
if (!this._conns[record.subscriptionUri].conn) {
this._conns[record.subscriptionUri].waitingForAlarm = false;
this._listenForMsgs(record.subscriptionUri);
}
}
},
// Start listening if subscriptions present.
_startConnectionsWaitingForAlarm: function() {
debug("startConnectionsWaitingForAlarm()");
for (let subscriptionUri in this._conns) {
if ((this._conns[subscriptionUri]) &&
!this._conns[subscriptionUri].conn &&
this._conns[subscriptionUri].waitingForAlarm) {
this._conns[subscriptionUri].waitingForAlarm = false;
this._listenForMsgs(subscriptionUri);
}
}
},
// Close connection and notify apps that subscription are gone.
_shutdownSubscription: function(aSubscriptionUri) {
debug("shutdownSubscriptions()");
if (typeof this._conns[aSubscriptionUri] == "object") {
if (this._conns[aSubscriptionUri].listener) {
this._conns[aSubscriptionUri].listener._pushService = null;
}
if (this._conns[aSubscriptionUri].channel) {
try {
this._conns[aSubscriptionUri].channel.cancel(Cr.NS_ERROR_ABORT);
} catch (e) {}
}
delete this._conns[aSubscriptionUri];
}
},
uninit: function() {
debug("uninit()");
this._shutdownConnections(true);
this._mainPushService = null;
},
request: function(action, aRecord) {
switch (action) {
case "register":
debug("register");
return this._subscribeResource(aRecord);
case "unregister":
this._shutdownSubscription(aRecord.subscriptionUri);
return this._unsubscribeResource(aRecord.subscriptionUri);
}
},
/** Push server has deleted subscription.
* Re-subscribe - if it succeeds send update db record and send
* pushsubscriptionchange,
* - on error delete record and send pushsubscriptionchange
* TODO: maybe pushsubscriptionerror will be included.
*/
_resubscribe: function(aSubscriptionUri) {
this._mainPushService.getByKeyID(aSubscriptionUri)
.then(record => this._subscribeResource(record)
.then(recordNew => {
if (this._mainPushService) {
this._mainPushService.updateRegistrationAndNotifyApp(aSubscriptionUri,
recordNew);
}
}, error => {
if (this._mainPushService) {
this._mainPushService.dropRegistrationAndNotifyApp(aSubscriptionUri);
}
})
);
},
connOnStop: function(aRequest, aSuccess,
aSubscriptionUri) {
debug("connOnStop() succeeded: " + aSuccess);
var conn = this._conns[aSubscriptionUri];
if (!conn) {
// there is no connection description that means that we closed
// connection, so do nothing. But we should have already deleted
// the listener.
return;
}
conn.channel = null;
conn.listener = null;
if (!aSuccess) {
this._retryAfterBackoff(aSubscriptionUri, -1);
} else if (Math.floor(aRequest.responseStatus / 100) == 5) {
var retryAfter = retryAfterParser(aRequest);
this._retryAfterBackoff(aSubscriptionUri, retryAfter);
} else if (Math.floor(aRequest.responseStatus / 100) == 4) {
this._shutdownSubscription(aSubscriptionUri);
this._resubscribe(aSubscriptionUri);
} else if (Math.floor(aRequest.responseStatus / 100) == 2) { // This should be 204
setTimeout(_ => this._listenForMsgs(aSubscriptionUri), 0);
} else {
this._retryAfterBackoff(aSubscriptionUri, -1);
}
},
_pushChannelOnStop: function(aUri, aAckUri, aMessage) {
debug("pushChannelOnStop() ");
let sendNotification = function(aAckUri, aPushRecord, self) {
aPushRecord.pushCount = aPushRecord.pushCount + 1;
aPushRecord.lastPush = new Date().getTime();
self._mainPushService.receivedPushMessage(aPushRecord, aMessage);
self._ackMsgRecv(aAckUri);
};
let recoverNoSuchEndpoint = function() {
debug("Could not get push endpoint " + aUri + " from DB");
};
this._mainPushService.getByKeyID(aUri)
.then(pushRecord => sendNotification(aAckUri, pushRecord, this),
recoverNoSuchEndpoint);
},
onAlarmFired: function() {
// Conditions are arranged in decreasing specificity.
// i.e. when _waitingForPong is true, other conditions are also true.
this._startConnectionsWaitingForAlarm();
},
prepareRegistration: function(aPushRecord) {
return {
pushEndpoint: aPushRecord.pushEndpoint,
pushReceiptEndpoint: aPushRecord.pushReceiptEndpoint,
version: aPushRecord.version,
lastPush: aPushRecord.lastPush,
pushCount: aPushRecord.pushCount
};
},
prepareRegister: function(aPushRecord) {
return {
pushEndpoint: aPushRecord.pushEndpoint,
pushReceiptEndpoint: aPushRecord.pushReceiptEndpoint
};
}
};

30
dom/push/PushServiceWebSocket.jsm
View File

@ -143,6 +143,10 @@ this.PushServiceWebSocket = {
objectStore.createIndex("scope", "scope", { unique: true });
},
getKeyFromRecord: function(aRecord) {
return aRecord.channelID;
},
newPushDB: function() {
return new PushDB(kPUSHWSDB_DB_NAME,
kPUSHWSDB_DB_VERSION,
@ -952,7 +956,7 @@ this.PushServiceWebSocket = {
var data = {channelID: record.channelID,
messageType: action};
return new Promise((resolve, reject) => {
var p = new Promise((resolve, reject) => {
this._pendingRequests[data.channelID] = {record: record,
resolve: resolve,
reject: reject,
@ -960,6 +964,11 @@ this.PushServiceWebSocket = {
};
this._queueRequest(data);
});
if (action == "unregister") {
return Promise.resolve();
} else {
return p;
}
},
_queueStart: Promise.resolve(),
@ -1043,7 +1052,7 @@ this.PushServiceWebSocket = {
debug("Could not get channelID " + aChannelIDFromServer + " from DB");
};
this._mainPushService.getByChannelID(aChannelID)
this._mainPushService.getByKeyID(aChannelID)
.then(compareRecordVersionAndNotify.bind(this),
err => recoverNoSuchChannelID(err));
},
@ -1096,7 +1105,7 @@ this.PushServiceWebSocket = {
};
}
this._mainPushService.getAllChannelIDs()
this._mainPushService.getAllKeyIDs()
.then(sendHelloMessage.bind(this),
sendHelloMessage.bind(this));
});
@ -1273,6 +1282,21 @@ this.PushServiceWebSocket = {
debug("UDP Server socket was shutdown. Status: " + aStatus);
this._udpServer = undefined;
this._beginWSSetup();
},
prepareRegistration: function(aPushRecord) {
return {
pushEndpoint: aPushRecord.pushEndpoint,
version: aPushRecord.version,
lastPush: aPushRecord.lastPush,
pushCount: aPushRecord.pushCount
};
},
prepareRegister: function(aPushRecord) {
return {
pushEndpoint: aPushRecord.pushEndpoint
};
}
};

1
dom/push/moz.build
View File

@ -12,6 +12,7 @@ EXTRA_COMPONENTS += [
EXTRA_PP_JS_MODULES += [
'PushDB.jsm',
'PushService.jsm',
'PushServiceHttp2.jsm',
'PushServiceWebSocket.jsm',
]

2
dom/push/test/xpcshell/test_clearAll_successful.js
View File

@ -42,6 +42,6 @@ add_task(function* test_unregister_success() {
});
yield PushNotificationService.clearAll();
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
ok(!record, 'Unregister did not remove record');
});

4
dom/push/test/xpcshell/test_notification_duplicate.js
View File

@ -72,11 +72,11 @@ add_task(function* test_notification_duplicate() {
yield waitForPromise(ackDefer.promise, DEFAULT_TIMEOUT,
'Timed out waiting for stale acknowledgement');
let staleRecord = yield db.getByChannelID(
let staleRecord = yield db.getByKeyID(
'8d2d9400-3597-4c5a-8a38-c546b0043bcc');
strictEqual(staleRecord.version, 2, 'Wrong stale record version');
let updatedRecord = yield db.getByChannelID(
let updatedRecord = yield db.getByKeyID(
'27d1e393-03ef-4c72-a5e6-9e890dfccad0');
strictEqual(updatedRecord.version, 3, 'Wrong updated record version');
});

2
dom/push/test/xpcshell/test_notification_error.js
View File

@ -56,7 +56,7 @@ add_task(function* test_notification_error() {
serverURI: "wss://push.example.org/",
networkInfo: new MockDesktopNetworkInfo(),
db: makeStub(db, {
getByChannelID(prev, channelID) {
getByKeyID(prev, channelID) {
if (channelID == '3c3930ba-44de-40dc-a7ca-8a133ec1a866') {
return Promise.reject('splines not reticulated');
}

2
dom/push/test/xpcshell/test_notification_incomplete.js
View File

@ -102,7 +102,7 @@ add_task(function* test_notification_incomplete() {
yield waitForPromise(notificationDefer.promise, DEFAULT_TIMEOUT,
'Timed out waiting for incomplete notifications');
let storeRecords = yield db.getAllChannelIDs();
let storeRecords = yield db.getAllKeyIDs();
storeRecords.sort(({pushEndpoint: a}, {pushEndpoint: b}) =>
compareAscending(a, b));
recordsAreEqual(records, storeRecords);

2
dom/push/test/xpcshell/test_notification_version_string.js
View File

@ -66,7 +66,7 @@ add_task(function* test_notification_version_string() {
yield waitForPromise(ackDefer.promise, DEFAULT_TIMEOUT,
'Timed out waiting for string acknowledgement');
let storeRecord = yield db.getByChannelID(
let storeRecord = yield db.getByKeyID(
'6ff97d56-d0c0-43bc-8f5b-61b855e1d93b');
strictEqual(storeRecord.version, 4, 'Wrong record version');
});

2
dom/push/test/xpcshell/test_register_case.js
View File

@ -56,7 +56,7 @@ add_task(function* test_register_case() {
equal(newRecord.scope, 'https://example.net/case',
'Wrong scope in registration record');
let record = yield db.getByChannelID(newRecord.channelID);
let record = yield db.getByKeyID(newRecord.channelID);
equal(record.pushEndpoint, 'https://example.com/update/case',
'Wrong push endpoint in database record');
equal(record.scope, 'https://example.net/case',

4
dom/push/test/xpcshell/test_register_flush.js
View File

@ -89,14 +89,14 @@ add_task(function* test_register_flush() {
yield waitForPromise(ackDefer.promise, DEFAULT_TIMEOUT,
'Timed out waiting for acknowledgements');
let prevRecord = yield db.getByChannelID(
let prevRecord = yield db.getByKeyID(
'9bcc7efb-86c7-4457-93ea-e24e6eb59b74');
equal(prevRecord.pushEndpoint, 'https://example.org/update/1',
'Wrong existing push endpoint');
strictEqual(prevRecord.version, 3,
'Should record version updates sent before register responses');
let registeredRecord = yield db.getByChannelID(newRecord.channelID);
let registeredRecord = yield db.getByKeyID(newRecord.channelID);
equal(registeredRecord.pushEndpoint, 'https://example.org/update/2',
'Wrong new push endpoint');
ok(!registeredRecord.version, 'Should not record premature updates');

2
dom/push/test/xpcshell/test_register_invalid_channel.js
View File

@ -55,6 +55,6 @@ add_task(function* test_register_invalid_channel() {
'Wrong error for invalid channel ID'
);
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
ok(!record, 'Should not store records for error responses');
});

2
dom/push/test/xpcshell/test_register_invalid_endpoint.js
View File

@ -57,6 +57,6 @@ add_task(function* test_register_invalid_endpoint() {
'Wrong error for invalid endpoint'
);
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
ok(!record, 'Should not store records with invalid endpoints');
});

2
dom/push/test/xpcshell/test_register_success.js
View File

@ -66,7 +66,7 @@ add_task(function* test_register_success() {
equal(newRecord.scope, 'https://example.org/1',
'Wrong scope in registration record');
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
equal(record.channelID, channelID,
'Wrong channel ID in database record');
equal(record.pushEndpoint, 'https://example.com/update/1',

2
dom/push/test/xpcshell/test_register_timeout.js
View File

@ -90,7 +90,7 @@ add_task(function* test_register_timeout() {
'Wrong error for request timeout'
);
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
ok(!record, 'Should not store records for timed-out responses');
yield waitForPromise(

2
dom/push/test/xpcshell/test_unregister_error.js
View File

@ -56,7 +56,7 @@ add_task(function* test_unregister_error() {
yield PushNotificationService.unregister(
'https://example.net/page/failure');
let result = yield db.getByChannelID(channelID);
let result = yield db.getByKeyID(channelID);
ok(!result, 'Deleted push record exists');
// Make sure we send a request to the server.

4
dom/push/test/xpcshell/test_unregister_invalid_json.js
View File

@ -62,13 +62,13 @@ add_task(function* test_unregister_invalid_json() {
// _sendRequest().
yield PushNotificationService.unregister(
'https://example.edu/page/1');
let record = yield db.getByChannelID(
let record = yield db.getByKeyID(
'87902e90-c57e-4d18-8354-013f4a556559');
ok(!record, 'Failed to delete unregistered record');
yield PushNotificationService.unregister(
'https://example.net/page/1');
record = yield db.getByChannelID(
record = yield db.getByKeyID(
'057caa8f-9b99-47ff-891c-adad18ce603e');
ok(!record,
'Failed to delete unregistered record after receiving invalid JSON');

2
dom/push/test/xpcshell/test_unregister_success.js
View File

@ -52,7 +52,7 @@ add_task(function* test_unregister_success() {
yield PushNotificationService.unregister(
'https://example.com/page/unregister-success');
let record = yield db.getByChannelID(channelID);
let record = yield db.getByKeyID(channelID);
ok(!record, 'Unregister did not remove record');
yield waitForPromise(unregisterDefer.promise, DEFAULT_TIMEOUT,

5
modules/libpref/init/all.js
View File

@ -4417,6 +4417,11 @@ pref("dom.push.adaptive.upperLimit", 1740000); // 29 min
// enable udp wakeup support
pref("dom.push.udp.wakeupEnabled", false);
// WebPush prefs:
pref("dom.push.http2.reset_retry_count_after_ms", 60000);
pref("dom.push.http2.maxRetries", 2);
pref("dom.push.http2.retryInterval", 5000);
// WebNetworkStats
pref("dom.mozNetworkStats.enabled", false);