upload handler, peer: add component to handling upload slots (close #25)

src/peer.js

@@ -5,7 +5,7 @@
 
 var BaseObject = require('base_object');
 var Storage = require('./storage');
-
+var UploadHandler = require('./upload_handler')
 
 class Peer extends BaseObject {
   initialize(params) {
@@ -14,6 +14,7 @@ class Peer extends BaseObject {
     this.swarm = params.swarm
     this.dataChannel = params.dataChannel
     this.dataChannel.on("data", (data) => this.messageReceived(data))
+    this.uploadHandler = new UploadHandler()
     this.active = true
     this.score = 1000
     this.sendPing()
@@ -38,23 +39,33 @@ class Peer extends BaseObject {
     this.dataChannel.send(message)
   }
 
+  sendSatisfy(resource) {
+    if (this.uploadHandler.getSlot(this.ident)) {
+      this.send('satisfy', resource, this.storage.getItem(resource))
+      this.swarm.chunksSent += 1
+    }
+  }
+
+  interestedReceived(resource) {
+    if (this.storage.contain(resource) && this.uploadHandler.getSlot(this.ident)) {
+      this.send("contain", resource)
+    } else {
+      this.send("choke", resource)
+    }
+  }
+
   messageReceived(data) {
     this.processMessage(data)
   }
 
   processMessage(data) {
     var [command, resource, content] = data.split("$")
     if (command === 'interested') {
-      if (this.storage.contain(resource)) {
-        this.send("contain", resource)
-      } else {
-        this.send("choke", resource)
-      }
+      this.interestedReceived(resource)
     } else if (command === "contain") {
       this.swarm.addSatisfyCandidate(this.ident, resource)
     } else if (command === 'request') {
-      this.send('satisfy', resource, this.storage.getItem(resource))
-      this.swarm.chunksSent += 1
+      this.sendSatisfy(resource)
     } else if (command === 'choke') {
       this.swarm.chokeReceived(resource)
     } else if (command === 'ping') {

src/settings.js

@@ -10,6 +10,21 @@ class Settings {
 Maximum size of the storage in bytes. */
 Settings.maxStorageChunks = 10
 
+/* maxPartners
+The maximum number of partners one peer can handle.
+Partners are used to be requested for video segments. */
+Settings.maxPartners = 4
+
+/* maxUploadSlots
+Maximum number of peers one can serve. */
+Settings.maxUploadSlots = 4
+
+/* uploadSlotTimeout
+Time in milliseconds that a upload slot will be expired.
+If a given downloader stops to request segments for 
+uploadSlotTimeout seconds, this slot will be emptied. */
+Settings.uploadSlotTimeout = 20000
+
 /* tracker
 Place where a rtc-switchboard server is running */
 Settings.tracker = 'http://server.bem.tv:8080'
@@ -25,12 +40,6 @@ will try to use P2P. Smaller than that, player will
 request chunks only for CDN. */
 Settings.lowBufferLength = 5
 
-/* maxPartners
-The maximum number of partners one peer can handle.
-Partners are used to be requested for video segments. */
-Settings.maxPartners = 4
-
-
 /* points
 How many points a partner win/loss when send a segment
 or not. This serves to reorganize peers and promoting or

src/upload_handler.js

@@ -0,0 +1,35 @@
+// Copyright 2014 Flávio Ribeiro <flavio@bem.tv>.
+// All rights reserved.
+// Use of this source code is governed by a Apache
+// license that can be found in the LICENSE file.
+
+var Settings = require("./settings")
+var _ = require("underscore")
+
+class UploadHandler {
+  constructor() {
+    this.maxUploadSlots = Settings.maxUploadSlots
+    this.slots = {}
+  }
+
+  getSlot(peerId) {
+    this.checkAndFreeSlots()
+    if (_.contains(this.slots.keys, peerId) || (_.size(this.slots) < this.maxUploadSlots)) {
+      this.slots[peerId] = Date.now()
+      return true
+    } else {
+      return false
+    }
+  }
+
+  checkAndFreeSlots() {
+    var now = Date.now() - Settings.uploadSlotTimeout
+    _.each(this.slots, function (timestamp, peerId) {
+      if (timestamp <= now) {
+        delete this.slots[peerId]
+      }
+    }, this)
+  }
+}
+
+module.exports = UploadHandler