dom/network/NetworkStatsService.jsm

   1 /* This Source Code Form is subject to the terms of the Mozilla Public
   2  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
   3  * You can obtain one at http://mozilla.org/MPL/2.0/. */
   4
   5 "use strict";
   6
   7 const DEBUG = false;
   8 function debug(s) {
   9   if (DEBUG) {
  10     dump("-*- NetworkStatsService: " + s + "\n");
  11   }
  12 }
  13
  14 const {classes: Cc, interfaces: Ci, utils: Cu, results: Cr} = Components;
  15
  16 this.EXPORTED_SYMBOLS = ["NetworkStatsService"];
  17
  18 Cu.import("resource://gre/modules/XPCOMUtils.jsm");
  19 Cu.import("resource://gre/modules/Services.jsm");
  20 Cu.import("resource://gre/modules/NetworkStatsDB.jsm");
  21 Cu.import("resource://gre/modules/Timer.jsm");
  22
  23 const NET_NETWORKSTATSSERVICE_CONTRACTID = "@mozilla.org/network/netstatsservice;1";
  24 const NET_NETWORKSTATSSERVICE_CID = Components.ID("{18725604-e9ac-488a-8aa0-2471e7f6c0a4}");
  25
  26 const TOPIC_BANDWIDTH_CONTROL = "netd-bandwidth-control"
  27
  28 const TOPIC_CONNECTION_STATE_CHANGED = "network-connection-state-changed";
  29 const NET_TYPE_WIFI = Ci.nsINetworkInterface.NETWORK_TYPE_WIFI;
  30 const NET_TYPE_MOBILE = Ci.nsINetworkInterface.NETWORK_TYPE_MOBILE;
  31
  32 // Networks have different status that NetworkStats API needs to be aware of.
  33 // Network is present and ready, so NetworkManager provides the whole info.
  34 const NETWORK_STATUS_READY   = 0;
  35 // Network is present but hasn't established a connection yet (e.g. SIM that has not
  36 // enabled 3G since boot).
  37 const NETWORK_STATUS_STANDBY = 1;
  38 // Network is not present, but stored in database by the previous connections.
  39 const NETWORK_STATUS_AWAY    = 2;
  40
  41 // The maximum traffic amount can be saved in the |cachedStats|.
  42 const MAX_CACHED_TRAFFIC = 500 * 1000 * 1000; // 500 MB
  43
  44 const QUEUE_TYPE_UPDATE_STATS = 0;
  45 const QUEUE_TYPE_UPDATE_CACHE = 1;
  46 const QUEUE_TYPE_WRITE_CACHE = 2;
  47
  48 XPCOMUtils.defineLazyServiceGetter(this, "ppmm",
  49                                    "@mozilla.org/parentprocessmessagemanager;1",
  50                                    "nsIMessageListenerManager");
  51
  52 XPCOMUtils.defineLazyServiceGetter(this, "gRil",
  53                                    "@mozilla.org/ril;1",
  54                                    "nsIRadioInterfaceLayer");
  55
  56 XPCOMUtils.defineLazyServiceGetter(this, "networkService",
  57                                    "@mozilla.org/network/service;1",
  58                                    "nsINetworkService");
  59
  60 XPCOMUtils.defineLazyServiceGetter(this, "appsService",
  61                                    "@mozilla.org/AppsService;1",
  62                                    "nsIAppsService");
  63
  64 XPCOMUtils.defineLazyServiceGetter(this, "gSettingsService",
  65                                    "@mozilla.org/settingsService;1",
  66                                    "nsISettingsService");
  67
  68 XPCOMUtils.defineLazyServiceGetter(this, "messenger",
  69                                    "@mozilla.org/system-message-internal;1",
  70                                    "nsISystemMessagesInternal");
  71
  72 this.NetworkStatsService = {
  73   init: function() {
  74     debug("Service started");
  75
  76     Services.obs.addObserver(this, "xpcom-shutdown", false);
  77     Services.obs.addObserver(this, TOPIC_CONNECTION_STATE_CHANGED, false);
  78     Services.obs.addObserver(this, TOPIC_BANDWIDTH_CONTROL, false);
  79     Services.obs.addObserver(this, "profile-after-change", false);
  80
  81     this.timer = Cc["@mozilla.org/timer;1"].createInstance(Ci.nsITimer);
  82
  83     // Object to store network interfaces, each network interface is composed
  84     // by a network object (network type and network Id) and a interfaceName
  85     // that contains the name of the physical interface (wlan0, rmnet0, etc.).
  86     // The network type can be 0 for wifi or 1 for mobile. On the other hand,
  87     // the network id is '0' for wifi or the iccid for mobile (SIM).
  88     // Each networkInterface is placed in the _networks object by the index of
  89     // 'networkId + networkType'.
  90     //
  91     // _networks object allows to map available network interfaces at low level
  92     // (wlan0, rmnet0, etc.) to a network. It's not mandatory to have a
  93     // networkInterface per network but can't exist a networkInterface not
  94     // being mapped to a network.
  95
  96     this._networks = Object.create(null);
  97
  98     // There is no way to know a priori if wifi connection is available,
  99     // just when the wifi driver is loaded, but it is unloaded when
 100     // wifi is switched off. So wifi connection is hardcoded
 101     let netId = this.getNetworkId('0', NET_TYPE_WIFI);
 102     this._networks[netId] = { network:       { id: '0',
 103                                                type: NET_TYPE_WIFI },
 104                               interfaceName: null,
 105                               status:        NETWORK_STATUS_STANDBY };
 106
 107     this.messages = ["NetworkStats:Get",
 108                      "NetworkStats:Clear",
 109                      "NetworkStats:ClearAll",
 110                      "NetworkStats:SetAlarm",
 111                      "NetworkStats:GetAlarms",
 112                      "NetworkStats:RemoveAlarms",
 113                      "NetworkStats:GetAvailableNetworks",
 114                      "NetworkStats:GetAvailableServiceTypes",
 115                      "NetworkStats:SampleRate",
 116                      "NetworkStats:MaxStorageAge"];
 117
 118     this.messages.forEach(function(aMsgName) {
 119       ppmm.addMessageListener(aMsgName, this);
 120     }, this);
 121
 122     this._db = new NetworkStatsDB();
 123
 124     // Stats for all interfaces are updated periodically
 125     this.timer.initWithCallback(this, this._db.sampleRate,
 126                                 Ci.nsITimer.TYPE_REPEATING_PRECISE);
 127
 128     // Stats not from netd are firstly stored in the cached.
 129     this.cachedStats = Object.create(null);
 130     this.cachedStatsDate = new Date();
 131
 132     this.updateQueue = [];
 133     this.isQueueRunning = false;
 134
 135     this._currentAlarms = {};
 136     this.initAlarms();
 137   },
 138
 139   receiveMessage: function(aMessage) {
 140     if (!aMessage.target.assertPermission("networkstats-manage")) {
 141       return;
 142     }
 143
 144     debug("receiveMessage " + aMessage.name);
 145
 146     let mm = aMessage.target;
 147     let msg = aMessage.json;
 148
 149     switch (aMessage.name) {
 150       case "NetworkStats:Get":
 151         this.getSamples(mm, msg);
 152         break;
 153       case "NetworkStats:Clear":
 154         this.clearInterfaceStats(mm, msg);
 155         break;
 156       case "NetworkStats:ClearAll":
 157         this.clearDB(mm, msg);
 158         break;
 159       case "NetworkStats:SetAlarm":
 160         this.setAlarm(mm, msg);
 161         break;
 162       case "NetworkStats:GetAlarms":
 163         this.getAlarms(mm, msg);
 164         break;
 165       case "NetworkStats:RemoveAlarms":
 166         this.removeAlarms(mm, msg);
 167         break;
 168       case "NetworkStats:GetAvailableNetworks":
 169         this.getAvailableNetworks(mm, msg);
 170         break;
 171       case "NetworkStats:GetAvailableServiceTypes":
 172         this.getAvailableServiceTypes(mm, msg);
 173         break;
 174       case "NetworkStats:SampleRate":
 175         // This message is sync.
 176         return this._db.sampleRate;
 177       case "NetworkStats:MaxStorageAge":
 178         // This message is sync.
 179         return this._db.maxStorageSamples * this._db.sampleRate;
 180     }
 181   },
 182
 183   observe: function observe(aSubject, aTopic, aData) {
 184     switch (aTopic) {
 185       case TOPIC_CONNECTION_STATE_CHANGED:
 186
 187         // If new interface is registered (notified from NetworkService),
 188         // the stats are updated for the new interface without waiting to
 189         // complete the updating period.
 190
 191         let network = aSubject.QueryInterface(Ci.nsINetworkInterface);
 192         debug("Network " + network.name + " of type " + network.type + " status change");
 193
 194         let netId = this.convertNetworkInterface(network);
 195         if (!netId) {
 196           break;
 197         }
 198
 199         this._updateCurrentAlarm(netId);
 200
 201         debug("NetId: " + netId);
 202         this.updateStats(netId);
 203         break;
 204
 205       case TOPIC_BANDWIDTH_CONTROL:
 206         debug("Bandwidth message from netd: " + JSON.stringify(aData));
 207
 208         let interfaceName = aData.substring(aData.lastIndexOf(" ") + 1);
 209         for (let networkId in this._networks) {
 210           if (interfaceName == this._networks[networkId].interfaceName) {
 211             let currentAlarm = this._currentAlarms[networkId];
 212             if (Object.getOwnPropertyNames(currentAlarm).length !== 0) {
 213               this._fireAlarm(currentAlarm.alarm);
 214             }
 215             break;
 216           }
 217         }
 218         break;
 219
 220       case "xpcom-shutdown":
 221         debug("Service shutdown");
 222
 223         this.messages.forEach(function(aMsgName) {
 224           ppmm.removeMessageListener(aMsgName, this);
 225         }, this);
 226
 227         Services.obs.removeObserver(this, "xpcom-shutdown");
 228         Services.obs.removeObserver(this, "profile-after-change");
 229         Services.obs.removeObserver(this, TOPIC_CONNECTION_STATE_CHANGED);
 230         Services.obs.removeObserver(this, TOPIC_BANDWIDTH_CONTROL);
 231
 232         this.timer.cancel();
 233         this.timer = null;
 234
 235         // Update stats before shutdown
 236         this.updateAllStats();
 237         break;
 238     }
 239   },
 240
 241   /*
 242    * nsITimerCallback
 243    * Timer triggers the update of all stats
 244    */
 245   notify: function(aTimer) {
 246     this.updateAllStats();
 247   },
 248
 249   /*
 250    * nsINetworkStatsService
 251    */
 252   getRilNetworks: function() {
 253     let networks = {};
 254     let numRadioInterfaces = gRil.numRadioInterfaces;
 255     for (let i = 0; i < numRadioInterfaces; i++) {
 256       let radioInterface = gRil.getRadioInterface(i);
 257       if (radioInterface.rilContext.iccInfo) {
 258         let netId = this.getNetworkId(radioInterface.rilContext.iccInfo.iccid,
 259                                       NET_TYPE_MOBILE);
 260         networks[netId] = { id : radioInterface.rilContext.iccInfo.iccid,
 261                             type: NET_TYPE_MOBILE };
 262       }
 263     }
 264     return networks;
 265   },
 266
 267   convertNetworkInterface: function(aNetwork) {
 268     if (aNetwork.type != NET_TYPE_MOBILE &&
 269         aNetwork.type != NET_TYPE_WIFI) {
 270       return null;
 271     }
 272
 273     let id = '0';
 274     if (aNetwork.type == NET_TYPE_MOBILE) {
 275       if (!(aNetwork instanceof Ci.nsIRilNetworkInterface)) {
 276         debug("Error! Mobile network should be an nsIRilNetworkInterface!");
 277         return null;
 278       }
 279
 280       let rilNetwork = aNetwork.QueryInterface(Ci.nsIRilNetworkInterface);
 281       id = rilNetwork.iccId;
 282     }
 283
 284     let netId = this.getNetworkId(id, aNetwork.type);
 285
 286     if (!this._networks[netId]) {
 287       this._networks[netId] = Object.create(null);
 288       this._networks[netId].network = { id: id,
 289                                         type: aNetwork.type };
 290     }
 291
 292     this._networks[netId].status = NETWORK_STATUS_READY;
 293     this._networks[netId].interfaceName = aNetwork.name;
 294     return netId;
 295   },
 296
 297   getNetworkId: function getNetworkId(aIccId, aNetworkType) {
 298     return aIccId + '' + aNetworkType;
 299   },
 300
 301   /* Function to ensure that one network is valid. The network is valid if its status is
 302    * NETWORK_STATUS_READY, NETWORK_STATUS_STANDBY or NETWORK_STATUS_AWAY.
 303    *
 304    * The result is |netId| or null in case of a non-valid network
 305    * aCallback is signatured as |function(netId)|.
 306    */
 307   validateNetwork: function validateNetwork(aNetwork, aCallback) {
 308     let netId = this.getNetworkId(aNetwork.id, aNetwork.type);
 309
 310     if (this._networks[netId]) {
 311       aCallback(netId);
 312       return;
 313     }
 314
 315     // Check if network is valid (RIL entry) but has not established a connection yet.
 316     // If so add to networks list with empty interfaceName.
 317     let rilNetworks = this.getRilNetworks();
 318     if (rilNetworks[netId]) {
 319       this._networks[netId] = Object.create(null);
 320       this._networks[netId].network = rilNetworks[netId];
 321       this._networks[netId].status = NETWORK_STATUS_STANDBY;
 322       this._currentAlarms[netId] = Object.create(null);
 323       aCallback(netId);
 324       return;
 325     }
 326
 327     // Check if network is available in the DB.
 328     this._db.isNetworkAvailable(aNetwork, function(aError, aResult) {
 329       if (aResult) {
 330         this._networks[netId] = Object.create(null);
 331         this._networks[netId].network = aNetwork;
 332         this._networks[netId].status = NETWORK_STATUS_AWAY;
 333         this._currentAlarms[netId] = Object.create(null);
 334         aCallback(netId);
 335         return;
 336       }
 337
 338       aCallback(null);
 339     }.bind(this));
 340   },
 341
 342   getAvailableNetworks: function getAvailableNetworks(mm, msg) {
 343     let self = this;
 344     let rilNetworks = this.getRilNetworks();
 345     this._db.getAvailableNetworks(function onGetNetworks(aError, aResult) {
 346
 347       // Also return the networks that are valid but have not
 348       // established connections yet.
 349       for (let netId in rilNetworks) {
 350         let found = false;
 351         for (let i = 0; i < aResult.length; i++) {
 352           if (netId == self.getNetworkId(aResult[i].id, aResult[i].type)) {
 353             found = true;
 354             break;
 355           }
 356         }
 357         if (!found) {
 358           aResult.push(rilNetworks[netId]);
 359         }
 360       }
 361
 362       mm.sendAsyncMessage("NetworkStats:GetAvailableNetworks:Return",
 363                           { id: msg.id, error: aError, result: aResult });
 364     });
 365   },
 366
 367   getAvailableServiceTypes: function getAvailableServiceTypes(mm, msg) {
 368     this._db.getAvailableServiceTypes(function onGetServiceTypes(aError, aResult) {
 369       mm.sendAsyncMessage("NetworkStats:GetAvailableServiceTypes:Return",
 370                           { id: msg.id, error: aError, result: aResult });
 371     });
 372   },
 373
 374   initAlarms: function initAlarms() {
 375     debug("Init usage alarms");
 376     let self = this;
 377
 378     for (let netId in this._networks) {
 379       this._currentAlarms[netId] = Object.create(null);
 380
 381       this._db.getFirstAlarm(netId, function getResult(error, result) {
 382         if (!error && result) {
 383           self._setAlarm(result, function onSet(error, success) {
 384             if (error == "InvalidStateError") {
 385               self._fireAlarm(result);
 386             }
 387           });
 388         }
 389       });
 390     }
 391   },
 392
 393   /*
 394    * Function called from manager to get stats from database.
 395    * In order to return updated stats, first is performed a call to
 396    * updateAllStats function, which will get last stats from netd
 397    * and update the database.
 398    * Then, depending on the request (stats per appId or total stats)
 399    * it retrieve them from database and return to the manager.
 400    */
 401   getSamples: function getSamples(mm, msg) {
 402     let network = msg.network;
 403     let netId = this.getNetworkId(network.id, network.type);
 404
 405     let appId = 0;
 406     let appManifestURL = msg.appManifestURL;
 407     if (appManifestURL) {
 408       appId = appsService.getAppLocalIdByManifestURL(appManifestURL);
 409
 410       if (!appId) {
 411         mm.sendAsyncMessage("NetworkStats:Get:Return",
 412                             { id: msg.id,
 413                               error: "Invalid appManifestURL", result: null });
 414         return;
 415       }
 416     }
 417
 418     let browsingTrafficOnly = msg.browsingTrafficOnly || false;
 419     let serviceType = msg.serviceType || "";
 420
 421     let start = new Date(msg.start);
 422     let end = new Date(msg.end);
 423
 424     let callback = (function (aError, aResult) {
 425       this._db.find(function onStatsFound(aError, aResult) {
 426         mm.sendAsyncMessage("NetworkStats:Get:Return",
 427                             { id: msg.id, error: aError, result: aResult });
 428       }, appId, browsingTrafficOnly, serviceType, network, start, end, appManifestURL);
 429     }).bind(this);
 430
 431     this.validateNetwork(network, function onValidateNetwork(aNetId) {
 432       if (!aNetId) {
 433         mm.sendAsyncMessage("NetworkStats:Get:Return",
 434                             { id: msg.id, error: "Invalid connectionType", result: null });
 435         return;
 436       }
 437
 438       // If network is currently active we need to update the cached stats first before
 439       // retrieving stats from the DB.
 440       if (this._networks[aNetId].status == NETWORK_STATUS_READY) {
 441         debug("getstats for network " + network.id + " of type " + network.type);
 442         debug("appId: " + appId + " from appManifestURL: " + appManifestURL);
 443         debug("browsingTrafficOnly: " + browsingTrafficOnly);
 444         debug("serviceType: " + serviceType);
 445
 446         if (appId || serviceType) {
 447           this.updateCachedStats(callback);
 448           return;
 449         }
 450
 451         this.updateStats(aNetId, function onStatsUpdated(aResult, aMessage) {
 452           this.updateCachedStats(callback);
 453         }.bind(this));
 454         return;
 455       }
 456
 457       // Network not active, so no need to update
 458       this._db.find(function onStatsFound(aError, aResult) {
 459         mm.sendAsyncMessage("NetworkStats:Get:Return",
 460                             { id: msg.id, error: aError, result: aResult });
 461       }, appId, browsingTrafficOnly, serviceType, network, start, end, appManifestURL);
 462     }.bind(this));
 463   },
 464
 465   clearInterfaceStats: function clearInterfaceStats(mm, msg) {
 466     let self = this;
 467     let network = msg.network;
 468
 469     debug("clear stats for network " + network.id + " of type " + network.type);
 470
 471     this.validateNetwork(network, function onValidateNetwork(aNetId) {
 472       if (!aNetId) {
 473         mm.sendAsyncMessage("NetworkStats:Clear:Return",
 474                             { id: msg.id, error: "Invalid connectionType", result: null });
 475         return;
 476       }
 477
 478       network = {network: network, networkId: aNetId};
 479       self.updateStats(aNetId, function onUpdate(aResult, aMessage) {
 480         if (!aResult) {
 481           mm.sendAsyncMessage("NetworkStats:Clear:Return",
 482                               { id: msg.id, error: aMessage, result: null });
 483           return;
 484         }
 485
 486         self._db.clearInterfaceStats(network, function onDBCleared(aError, aResult) {
 487           self._updateCurrentAlarm(aNetId);
 488           mm.sendAsyncMessage("NetworkStats:Clear:Return",
 489                               { id: msg.id, error: aError, result: aResult });
 490         });
 491       });
 492     });
 493   },
 494
 495   clearDB: function clearDB(mm, msg) {
 496     let self = this;
 497     this._db.getAvailableNetworks(function onGetNetworks(aError, aResult) {
 498       if (aError) {
 499         mm.sendAsyncMessage("NetworkStats:ClearAll:Return",
 500                             { id: msg.id, error: aError, result: aResult });
 501         return;
 502       }
 503
 504       let networks = aResult;
 505       networks.forEach(function(network, index) {
 506         networks[index] = {network: network, networkId: self.getNetworkId(network.id, network.type)};
 507       }, self);
 508
 509       self.updateAllStats(function onUpdate(aResult, aMessage){
 510         if (!aResult) {
 511           mm.sendAsyncMessage("NetworkStats:ClearAll:Return",
 512                               { id: msg.id, error: aMessage, result: null });
 513           return;
 514         }
 515
 516         self._db.clearStats(networks, function onDBCleared(aError, aResult) {
 517           networks.forEach(function(network, index) {
 518             self._updateCurrentAlarm(network.networkId);
 519           }, self);
 520           mm.sendAsyncMessage("NetworkStats:ClearAll:Return",
 521                               { id: msg.id, error: aError, result: aResult });
 522         });
 523       });
 524     });
 525   },
 526
 527   updateAllStats: function updateAllStats(aCallback) {
 528     let elements = [];
 529     let lastElement = null;
 530     let callback = (function (success, message) {
 531       this.updateCachedStats(aCallback);
 532     }).bind(this);
 533
 534     // For each connectionType create an object containning the type
 535     // and the 'queueIndex', the 'queueIndex' is an integer representing
 536     // the index of a connection type in the global queue array. So, if
 537     // the connection type is already in the queue it is not appended again,
 538     // else it is pushed in 'elements' array, which later will be pushed to
 539     // the queue array.
 540     for (let netId in this._networks) {
 541       if (this._networks[netId].status != NETWORK_STATUS_READY) {
 542         continue;
 543       }
 544
 545       lastElement = { netId: netId,
 546                       queueIndex: this.updateQueueIndex(netId) };
 547
 548       if (lastElement.queueIndex == -1) {
 549         elements.push({ netId:     lastElement.netId,
 550                         callbacks: [],
 551                         queueType: QUEUE_TYPE_UPDATE_STATS });
 552       }
 553     }
 554
 555     if (!lastElement) {
 556       // No elements need to be updated, probably because status is different than
 557       // NETWORK_STATUS_READY.
 558       if (aCallback) {
 559         aCallback(true, "OK");
 560       }
 561       return;
 562     }
 563
 564     if (elements.length > 0) {
 565       // If length of elements is greater than 0, callback is set to
 566       // the last element.
 567       elements[elements.length - 1].callbacks.push(callback);
 568       this.updateQueue = this.updateQueue.concat(elements);
 569     } else {
 570       // Else, it means that all connection types are already in the queue to
 571       // be updated, so callback for this request is added to
 572       // the element in the main queue with the index of the last 'lastElement'.
 573       // But before is checked that element is still in the queue because it can
 574       // be processed while generating 'elements' array.
 575       let element = this.updateQueue[lastElement.queueIndex];
 576       if (aCallback &&
 577          (!element || element.netId != lastElement.netId)) {
 578         aCallback();
 579         return;
 580       }
 581
 582       this.updateQueue[lastElement.queueIndex].callbacks.push(callback);
 583     }
 584
 585     // Call the function that process the elements of the queue.
 586     this.processQueue();
 587
 588     if (DEBUG) {
 589       this.logAllRecords();
 590     }
 591   },
 592
 593   updateStats: function updateStats(aNetId, aCallback) {
 594     // Check if the connection is in the main queue, push a new element
 595     // if it is not being processed or add a callback if it is.
 596     let index = this.updateQueueIndex(aNetId);
 597     if (index == -1) {
 598       this.updateQueue.push({ netId: aNetId,
 599                               callbacks: [aCallback],
 600                               queueType: QUEUE_TYPE_UPDATE_STATS });
 601     } else {
 602       this.updateQueue[index].callbacks.push(aCallback);
 603       return;
 604     }
 605
 606     // Call the function that process the elements of the queue.
 607     this.processQueue();
 608   },
 609
 610   /*
 611    * Find if a connection is in the main queue array and return its
 612    * index, if it is not in the array return -1.
 613    */
 614   updateQueueIndex: function updateQueueIndex(aNetId) {
 615     return this.updateQueue.map(function(e) { return e.netId; }).indexOf(aNetId);
 616   },
 617
 618   /*
 619    * Function responsible of process all requests in the queue.
 620    */
 621   processQueue: function processQueue(aResult, aMessage) {
 622     // If aResult is not undefined, the caller of the function is the result
 623     // of processing an element, so remove that element and call the callbacks
 624     // it has.
 625     let self = this;
 626
 627     if (aResult != undefined) {
 628       let item = this.updateQueue.shift();
 629       for (let callback of item.callbacks) {
 630         if (callback) {
 631           callback(aResult, aMessage);
 632         }
 633       }
 634     } else {
 635       // The caller is a function that has pushed new elements to the queue,
 636       // if isQueueRunning is false it means there is no processing currently
 637       // being done, so start.
 638       if (this.isQueueRunning) {
 639         return;
 640       } else {
 641         this.isQueueRunning = true;
 642       }
 643     }
 644
 645     // Check length to determine if queue is empty and stop processing.
 646     if (this.updateQueue.length < 1) {
 647       this.isQueueRunning = false;
 648       return;
 649     }
 650
 651     // Process the next item as soon as possible.
 652     setTimeout(function () {
 653                  self.run(self.updateQueue[0]);
 654                }, 0);
 655   },
 656
 657   run: function run(item) {
 658     switch (item.queueType) {
 659       case QUEUE_TYPE_UPDATE_STATS:
 660         this.update(item.netId, this.processQueue.bind(this));
 661         break;
 662       case QUEUE_TYPE_UPDATE_CACHE:
 663         this.updateCache(this.processQueue.bind(this));
 664         break;
 665       case QUEUE_TYPE_WRITE_CACHE:
 666         this.writeCache(item.stats, this.processQueue.bind(this));
 667         break;
 668     }
 669   },
 670
 671   update: function update(aNetId, aCallback) {
 672     // Check if connection type is valid.
 673     if (!this._networks[aNetId]) {
 674       if (aCallback) {
 675         aCallback(false, "Invalid network " + aNetId);
 676       }
 677       return;
 678     }
 679
 680     let interfaceName = this._networks[aNetId].interfaceName;
 681     debug("Update stats for " + interfaceName);
 682
 683     // Request stats to NetworkService, which will get stats from netd, passing
 684     // 'networkStatsAvailable' as a callback.
 685     if (interfaceName) {
 686       networkService.getNetworkInterfaceStats(interfaceName,
 687                 this.networkStatsAvailable.bind(this, aCallback, aNetId));
 688       return;
 689     }
 690
 691     if (aCallback) {
 692       aCallback(true, "ok");
 693     }
 694   },
 695
 696   /*
 697    * Callback of request stats. Store stats in database.
 698    */
 699   networkStatsAvailable: function networkStatsAvailable(aCallback, aNetId,
 700                                                         aResult, aRxBytes,
 701                                                         aTxBytes, aTimestamp) {
 702     if (!aResult) {
 703       if (aCallback) {
 704         aCallback(false, "Netd IPC error");
 705       }
 706       return;
 707     }
 708
 709     let stats = { appId:          0,
 710                   isInBrowser:    false,
 711                   serviceType:    "",
 712                   networkId:      this._networks[aNetId].network.id,
 713                   networkType:    this._networks[aNetId].network.type,
 714                   date:           new Date(aTimestamp),
 715                   rxBytes:        aTxBytes,
 716                   txBytes:        aRxBytes,
 717                   isAccumulative: true };
 718
 719     debug("Update stats for: " + JSON.stringify(stats));
 720
 721     this._db.saveStats(stats, function onSavedStats(aError, aResult) {
 722       if (aCallback) {
 723         if (aError) {
 724           aCallback(false, aError);
 725           return;
 726         }
 727
 728         aCallback(true, "OK");
 729       }
 730     });
 731   },
 732
 733   /*
 734    * Function responsible for receiving stats which are not from netd.
 735    */
 736   saveStats: function saveStats(aAppId, aIsInBrowser, aServiceType, aNetwork,
 737                                 aTimeStamp, aRxBytes, aTxBytes, aIsAccumulative,
 738                                 aCallback) {
 739     let netId = this.convertNetworkInterface(aNetwork);
 740     if (!netId) {
 741       if (aCallback) {
 742         aCallback(false, "Invalid network type");
 743       }
 744       return;
 745     }
 746
 747     // Check if |aConnectionType|, |aAppId| and |aServiceType| are valid.
 748     // There are two invalid cases for the combination of |aAppId| and
 749     // |aServiceType|:
 750     // a. Both |aAppId| is non-zero and |aServiceType| is non-empty.
 751     // b. Both |aAppId| is zero and |aServiceType| is empty.
 752     if (!this._networks[netId] || (aAppId && aServiceType) ||
 753         (!aAppId && !aServiceType)) {
 754       debug("Invalid network interface, appId or serviceType");
 755       return;
 756     }
 757
 758     let stats = { appId:          aAppId,
 759                   isInBrowser:    aIsInBrowser,
 760                   serviceType:    aServiceType,
 761                   networkId:      this._networks[netId].network.id,
 762                   networkType:    this._networks[netId].network.type,
 763                   date:           new Date(aTimeStamp),
 764                   rxBytes:        aRxBytes,
 765                   txBytes:        aTxBytes,
 766                   isAccumulative: aIsAccumulative };
 767
 768     this.updateQueue.push({ stats: stats,
 769                             callbacks: [aCallback],
 770                             queueType: QUEUE_TYPE_WRITE_CACHE });
 771
 772     this.processQueue();
 773   },
 774
 775   /*
 776    *
 777    */
 778   writeCache: function writeCache(aStats, aCallback) {
 779     debug("saveStats: " + aStats.appId + " " + aStats.isInBrowser + " " +
 780           aStats.serviceType + " " + aStats.networkId + " " +
 781           aStats.networkType + " " + aStats.date + " " +
 782           aStats.rxBytes + " " + aStats.txBytes);
 783
 784     // Generate an unique key from |appId|, |isInBrowser|, |serviceType| and
 785     // |netId|, which is used to retrieve data in |cachedStats|.
 786     let netId = this.getNetworkId(aStats.networkId, aStats.networkType);
 787     let key = aStats.appId + "" + aStats.isInBrowser + "" +
 788               aStats.serviceType + "" + netId;
 789
 790     // |cachedStats| only keeps the data with the same date.
 791     // If the incoming date is different from |cachedStatsDate|,
 792     // both |cachedStats| and |cachedStatsDate| will get updated.
 793     let diff = (this._db.normalizeDate(aStats.date) -
 794                 this._db.normalizeDate(this.cachedStatsDate)) /
 795                this._db.sampleRate;
 796     if (diff != 0) {
 797       this.updateCache(function onUpdated(success, message) {
 798         this.cachedStatsDate = aStats.date;
 799         this.cachedStats[key] = aStats;
 800
 801         if (aCallback) {
 802           aCallback(true, "ok");
 803         }
 804       }.bind(this));
 805       return;
 806     }
 807
 808     // Try to find the matched row in the cached by |appId| and |connectionType|.
 809     // If not found, save the incoming data into the cached.
 810     let cachedStats = this.cachedStats[key];
 811     if (!cachedStats) {
 812       this.cachedStats[key] = aStats;
 813       if (aCallback) {
 814         aCallback(true, "ok");
 815       }
 816       return;
 817     }
 818
 819     // Find matched row, accumulate the traffic amount.
 820     cachedStats.rxBytes += aStats.rxBytes;
 821     cachedStats.txBytes += aStats.txBytes;
 822
 823     // If new rxBytes or txBytes exceeds MAX_CACHED_TRAFFIC
 824     // the corresponding row will be saved to indexedDB.
 825     // Then, the row will be removed from the cached.
 826     if (cachedStats.rxBytes > MAX_CACHED_TRAFFIC ||
 827         cachedStats.txBytes > MAX_CACHED_TRAFFIC) {
 828       this._db.saveStats(cachedStats, function (error, result) {
 829         debug("Application stats inserted in indexedDB");
 830         if (aCallback) {
 831           aCallback(true, "ok");
 832         }
 833       });
 834       delete this.cachedStats[key];
 835       return;
 836     }
 837
 838     if (aCallback) {
 839       aCallback(true, "ok");
 840     }
 841   },
 842
 843   updateCachedStats: function updateCachedStats(aCallback) {
 844     this.updateQueue.push({ callbacks: [aCallback],
 845                             queueType: QUEUE_TYPE_UPDATE_CACHE });
 846
 847     this.processQueue();
 848   },
 849
 850   updateCache: function updateCache(aCallback) {
 851     debug("updateCache: " + this.cachedStatsDate);
 852
 853     let stats = Object.keys(this.cachedStats);
 854     if (stats.length == 0) {
 855       // |cachedStats| is empty, no need to update.
 856       if (aCallback) {
 857         aCallback(true, "no need to update");
 858       }
 859       return;
 860     }
 861
 862     let index = 0;
 863     this._db.saveStats(this.cachedStats[stats[index]],
 864                        function onSavedStats(error, result) {
 865       debug("Application stats inserted in indexedDB");
 866
 867       // Clean up the |cachedStats| after updating.
 868       if (index == stats.length - 1) {
 869         this.cachedStats = Object.create(null);
 870
 871         if (aCallback) {
 872           aCallback(true, "ok");
 873         }
 874         return;
 875       }
 876
 877       // Update is not finished, keep updating.
 878       index += 1;
 879       this._db.saveStats(this.cachedStats[stats[index]],
 880                          onSavedStats.bind(this, error, result));
 881     }.bind(this));
 882   },
 883
 884   get maxCachedTraffic () {
 885     return MAX_CACHED_TRAFFIC;
 886   },
 887
 888   logAllRecords: function logAllRecords() {
 889     this._db.logAllRecords(function onResult(aError, aResult) {
 890       if (aError) {
 891         debug("Error: " + aError);
 892         return;
 893       }
 894
 895       debug("===== LOG =====");
 896       debug("There are " + aResult.length + " items");
 897       debug(JSON.stringify(aResult));
 898     });
 899   },
 900
 901   getAlarms: function getAlarms(mm, msg) {
 902     let self = this;
 903     let network = msg.data.network;
 904     let manifestURL = msg.data.manifestURL;
 905
 906     if (network) {
 907       this.validateNetwork(network, function onValidateNetwork(aNetId) {
 908         if (!aNetId) {
 909           mm.sendAsyncMessage("NetworkStats:GetAlarms:Return",
 910                               { id: msg.id, error: "InvalidInterface", result: null });
 911           return;
 912         }
 913
 914         self._getAlarms(mm, msg, aNetId, manifestURL);
 915       });
 916       return;
 917     }
 918
 919     this._getAlarms(mm, msg, null, manifestURL);
 920   },
 921
 922   _getAlarms: function _getAlarms(mm, msg, aNetId, aManifestURL) {
 923     let self = this;
 924     this._db.getAlarms(aNetId, aManifestURL, function onCompleted(error, result) {
 925       if (error) {
 926         mm.sendAsyncMessage("NetworkStats:GetAlarms:Return",
 927                             { id: msg.id, error: error, result: result });
 928         return;
 929       }
 930
 931       let alarms = []
 932       // NetworkStatsManager must return the network instead of the networkId.
 933       for (let i = 0; i < result.length; i++) {
 934         let alarm = result[i];
 935         alarms.push({ id: alarm.id,
 936                       network: self._networks[alarm.networkId].network,
 937                       threshold: alarm.absoluteThreshold,
 938                       data: alarm.data });
 939       }
 940
 941       mm.sendAsyncMessage("NetworkStats:GetAlarms:Return",
 942                           { id: msg.id, error: null, result: alarms });
 943     });
 944   },
 945
 946   removeAlarms: function removeAlarms(mm, msg) {
 947     let alarmId = msg.data.alarmId;
 948     let manifestURL = msg.data.manifestURL;
 949
 950     let self = this;
 951     let callback = function onRemove(error, result) {
 952       if (error) {
 953         mm.sendAsyncMessage("NetworkStats:RemoveAlarms:Return",
 954                             { id: msg.id, error: error, result: result });
 955         return;
 956       }
 957
 958       for (let i in self._currentAlarms) {
 959         let currentAlarm = self._currentAlarms[i].alarm;
 960         if (currentAlarm && ((alarmId == currentAlarm.id) ||
 961             (alarmId == -1 && currentAlarm.manifestURL == manifestURL))) {
 962
 963           self._updateCurrentAlarm(currentAlarm.networkId);
 964         }
 965       }
 966
 967       mm.sendAsyncMessage("NetworkStats:RemoveAlarms:Return",
 968                           { id: msg.id, error: error, result: true });
 969     };
 970
 971     if (alarmId == -1) {
 972       this._db.removeAlarms(manifestURL, callback);
 973     } else {
 974       this._db.removeAlarm(alarmId, manifestURL, callback);
 975     }
 976   },
 977
 978   /*
 979    * Function called from manager to set an alarm.
 980    */
 981   setAlarm: function setAlarm(mm, msg) {
 982     let options = msg.data;
 983     let network = options.network;
 984     let threshold = options.threshold;
 985
 986     debug("Set alarm at " + threshold + " for " + JSON.stringify(network));
 987
 988     if (threshold < 0) {
 989       mm.sendAsyncMessage("NetworkStats:SetAlarm:Return",
 990                           { id: msg.id, error: "InvalidThresholdValue", result: null });
 991       return;
 992     }
 993
 994     let self = this;
 995     this.validateNetwork(network, function onValidateNetwork(aNetId) {
 996       if (!aNetId) {
 997         mm.sendAsyncMessage("NetworkStats:SetAlarm:Return",
 998                             { id: msg.id, error: "InvalidiConnectionType", result: null });
 999         return;
1000       }
1001
1002       let newAlarm = {
1003         id: null,
1004         networkId: aNetId,
1005         absoluteThreshold: threshold,
1006         relativeThreshold: null,
1007         startTime: options.startTime,
1008         data: options.data,
1009         pageURL: options.pageURL,
1010         manifestURL: options.manifestURL
1011       };
1012
1013       self._getAlarmQuota(newAlarm, function onUpdate(error, quota) {
1014         if (error) {
1015           mm.sendAsyncMessage("NetworkStats:SetAlarm:Return",
1016                               { id: msg.id, error: error, result: null });
1017           return;
1018         }
1019
1020         self._db.addAlarm(newAlarm, function addSuccessCb(error, newId) {
1021           if (error) {
1022             mm.sendAsyncMessage("NetworkStats:SetAlarm:Return",
1023                                 { id: msg.id, error: error, result: null });
1024             return;
1025           }
1026
1027           newAlarm.id = newId;
1028           self._setAlarm(newAlarm, function onSet(error, success) {
1029             mm.sendAsyncMessage("NetworkStats:SetAlarm:Return",
1030                                 { id: msg.id, error: error, result: newId });
1031
1032             if (error == "InvalidStateError") {
1033               self._fireAlarm(newAlarm);
1034             }
1035           });
1036         });
1037       });
1038     });
1039   },
1040
1041   _setAlarm: function _setAlarm(aAlarm, aCallback) {
1042     let currentAlarm = this._currentAlarms[aAlarm.networkId];
1043     if ((Object.getOwnPropertyNames(currentAlarm).length !== 0 &&
1044          aAlarm.relativeThreshold > currentAlarm.alarm.relativeThreshold) ||
1045         this._networks[aAlarm.networkId].status != NETWORK_STATUS_READY) {
1046       aCallback(null, true);
1047       return;
1048     }
1049
1050     let self = this;
1051
1052     this._getAlarmQuota(aAlarm, function onUpdate(aError, aQuota) {
1053       if (aError) {
1054         aCallback(aError, null);
1055         return;
1056       }
1057
1058       let callback = function onAlarmSet(aError) {
1059         if (aError) {
1060           debug("Set alarm error: " + aError);
1061           aCallback("netdError", null);
1062           return;
1063         }
1064
1065         self._currentAlarms[aAlarm.networkId].alarm = aAlarm;
1066
1067         aCallback(null, true);
1068       };
1069
1070       debug("Set alarm " + JSON.stringify(aAlarm));
1071       let interfaceName = self._networks[aAlarm.networkId].interfaceName;
1072       if (interfaceName) {
1073         networkService.setNetworkInterfaceAlarm(interfaceName,
1074                                                 aQuota,
1075                                                 callback);
1076         return;
1077       }
1078
1079       aCallback(null, true);
1080     });
1081   },
1082
1083   _getAlarmQuota: function _getAlarmQuota(aAlarm, aCallback) {
1084     let self = this;
1085     this.updateStats(aAlarm.networkId, function onStatsUpdated(aResult, aMessage) {
1086       self._db.getCurrentStats(self._networks[aAlarm.networkId].network,
1087                                aAlarm.startTime,
1088                                function onStatsFound(error, result) {
1089         if (error) {
1090           debug("Error getting stats for " +
1091                 JSON.stringify(self._networks[aAlarm.networkId]) + ": " + error);
1092           aCallback(error, result);
1093           return;
1094         }
1095
1096         let quota = aAlarm.absoluteThreshold - result.rxBytes - result.txBytes;
1097
1098         // Alarm set to a threshold lower than current rx/tx bytes.
1099         if (quota <= 0) {
1100           aCallback("InvalidStateError", null);
1101           return;
1102         }
1103
1104         aAlarm.relativeThreshold = aAlarm.startTime
1105                                  ? result.rxTotalBytes + result.txTotalBytes + quota
1106                                  : aAlarm.absoluteThreshold;
1107
1108         aCallback(null, quota);
1109       });
1110     });
1111   },
1112
1113   _fireAlarm: function _fireAlarm(aAlarm) {
1114     debug("Fire alarm");
1115
1116     let self = this;
1117     this._db.removeAlarm(aAlarm.id, null, function onRemove(aError, aResult){
1118       if (!aError && !aResult) {
1119         return;
1120       }
1121
1122       self._fireSystemMessage(aAlarm);
1123       self._updateCurrentAlarm(aAlarm.networkId);
1124     });
1125   },
1126
1127   _updateCurrentAlarm: function _updateCurrentAlarm(aNetworkId) {
1128     this._currentAlarms[aNetworkId] = Object.create(null);
1129
1130     let self = this;
1131     this._db.getFirstAlarm(aNetworkId, function onGet(error, result){
1132       if (error) {
1133         debug("Error getting the first alarm");
1134         return;
1135       }
1136
1137       if (!result) {
1138         let interfaceName = self._networks[aNetworkId].interfaceName;
1139         networkService.setNetworkInterfaceAlarm(interfaceName, -1,
1140                                                 function onComplete(){});
1141         return;
1142       }
1143
1144       self._setAlarm(result, function onSet(error, success){
1145         if (error == "InvalidStateError") {
1146           self._fireAlarm(result);
1147           return;
1148         }
1149       });
1150     });
1151   },
1152
1153   _fireSystemMessage: function _fireSystemMessage(aAlarm) {
1154     debug("Fire system message: " + JSON.stringify(aAlarm));
1155
1156     let manifestURI = Services.io.newURI(aAlarm.manifestURL, null, null);
1157     let pageURI = Services.io.newURI(aAlarm.pageURL, null, null);
1158
1159     let alarm = { "id":        aAlarm.id,
1160                   "threshold": aAlarm.absoluteThreshold,
1161                   "data":      aAlarm.data };
1162     messenger.sendMessage("networkstats-alarm", alarm, pageURI, manifestURI);
1163   }
1164 };
1165
1166 NetworkStatsService.init();