feat(streams): Add multiple upstreams for basic load balancing

backend/models/stream.js

@@ -1,55 +1,65 @@
 // Objection Docs:
 // http://vincit.github.io/objection.js/
 
-const db    = require('../db');
+const db = require('../db');
 const Model = require('objection').Model;
-const User  = require('./user');
-const now   = require('./now_helper');
+const User = require('./user');
+const now = require('./now_helper');
 
 Model.knex(db);
 
 class Stream extends Model {
-	$beforeInsert () {
-		this.created_on  = now();
-		this.modified_on = now();
+  $beforeInsert() {
+    this.created_on = now();
+    this.modified_on = now();
 
-		// Default for meta
-		if (typeof this.meta === 'undefined') {
-			this.meta = {};
-		}
-	}
+    // Default for forwarding_hosts
+    if (typeof this.forwarding_hosts === 'undefined') {
+      this.forwarding_hosts = [];
+    }
 
-	$beforeUpdate () {
-		this.modified_on = now();
-	}
+    // Default for meta
+    if (typeof this.meta === 'undefined') {
+      this.meta = {};
+    }
+  }
 
-	static get name () {
-		return 'Stream';
-	}
+  $beforeUpdate() {
+    this.modified_on = now();
 
-	static get tableName () {
-		return 'stream';
-	}
+    // Sort domain_names
+    if (typeof this.forwarding_hosts !== 'undefined') {
+      this.forwarding_hosts.sort();
+    }
+  }
 
-	static get jsonAttributes () {
-		return ['meta'];
-	}
+  static get name() {
+    return 'Stream';
+  }
 
-	static get relationMappings () {
-		return {
-			owner: {
-				relation:   Model.HasOneRelation,
-				modelClass: User,
-				join:       {
-					from: 'stream.owner_user_id',
-					to:   'user.id'
-				},
-				modify: function (qb) {
-					qb.where('user.is_deleted', 0);
-				}
-			}
-		};
-	}
+  static get tableName() {
+    return 'stream';
+  }
+
+  static get jsonAttributes() {
+    return ['forwarding_hosts', 'meta'];
+  }
+
+  static get relationMappings() {
+    return {
+      owner: {
+        relation: Model.HasOneRelation,
+        modelClass: User,
+        join: {
+          from: 'stream.owner_user_id',
+          to: 'user.id'
+        },
+        modify: function (qb) {
+          qb.where('user.is_deleted', 0);
+        }
+      }
+    };
+  }
 }
 
 module.exports = Stream;

backend/schema/endpoints/streams.json

@@ -20,7 +20,7 @@
       "minimum": 1,
       "maximum": 65535
     },
-    "forwarding_host": {
+    "host": {
       "anyOf": [
         {
           "$ref": "../definitions.json#/definitions/domain_name"
@@ -35,6 +35,22 @@
         }
       ]
     },
+    "forwarding_hosts": {
+      "anyOf": [
+        {
+          "$ref": "#/definitions/host"
+        },
+        {
+          "type": "array",
+          "minItems": 1,
+          "maxItems": 15,
+          "uniqueItems": true,
+          "items": {
+            "$ref": "#/definitions/host"
+          }
+        }
+      ]
+    },
     "forwarding_port": {
       "type": "integer",
       "minimum": 1,
@@ -66,8 +82,8 @@
     "incoming_port": {
       "$ref": "#/definitions/incoming_port"
     },
-    "forwarding_host": {
-      "$ref": "#/definitions/forwarding_host"
+    "forwarding_hosts": {
+      "$ref": "#/definitions/forwarding_hosts"
     },
     "forwarding_port": {
       "$ref": "#/definitions/forwarding_port"
@@ -118,15 +134,15 @@
         "additionalProperties": false,
         "required": [
           "incoming_port",
-          "forwarding_host",
+          "forwarding_hosts",
           "forwarding_port"
         ],
         "properties": {
           "incoming_port": {
             "$ref": "#/definitions/incoming_port"
           },
-          "forwarding_host": {
-            "$ref": "#/definitions/forwarding_host"
+          "forwarding_hosts": {
+            "$ref": "#/definitions/forwarding_hosts"
           },
           "forwarding_port": {
             "$ref": "#/definitions/forwarding_port"
@@ -165,8 +181,8 @@
           "incoming_port": {
             "$ref": "#/definitions/incoming_port"
           },
-          "forwarding_host": {
-            "$ref": "#/definitions/forwarding_host"
+          "forwarding_hosts": {
+            "$ref": "#/definitions/forwarding_hosts"
           },
           "forwarding_port": {
             "$ref": "#/definitions/forwarding_port"
@@ -231,4 +247,4 @@
       }
     }
   ]
-}
+}

backend/templates/stream.conf

@@ -3,6 +3,13 @@
 # ------------------------------------------------------------
 
 {% if enabled %}
+
+upstream stream_{{ incoming_port }}_tcp {
+  {% for forwarding_host in forwarding_hosts %}
+  server {{ forwarding_host }}:{{ forwarding_port }};
+  {%- endfor %}
+}
+
 {% if tcp_forwarding == 1 or tcp_forwarding == true -%}
 server {
   listen {{ incoming_port }};
@@ -12,7 +19,7 @@ server {
   #listen [::]:{{ incoming_port }};
 {% endif %}
 
-  proxy_pass {{ forwarding_host }}:{{ forwarding_port }};
+  proxy_pass stream_{{ incoming_port }}_tcp;
 
   # Custom
   include /data/nginx/custom/server_stream[.]conf;
@@ -20,18 +27,26 @@ server {
 }
 {% endif %}
 {% if udp_forwarding == 1 or udp_forwarding == true %}
+
+upstream stream_{{ incoming_port }}_udp {
+  {% for forwarding_host in forwarding_hosts %}
+  server {{ forwarding_host }}:{{ forwarding_port }};
+  {%- endfor %}
+}
+
 server {
   listen {{ incoming_port }} udp;
 {% if ipv6 -%}
   listen [::]:{{ incoming_port }} udp;
 {% else -%}
-  #listen [::]:{{ incoming_port }} udp;
+   #listen [::]:{{ incoming_port }} udp;
 {% endif %}
-  proxy_pass {{ forwarding_host }}:{{ forwarding_port }};
+
+  proxy_pass stream_{{ incoming_port }}_udp;
 
   # Custom
   include /data/nginx/custom/server_stream[.]conf;
   include /data/nginx/custom/server_stream_udp[.]conf;
 }
 {% endif %}
-{% endif %}
+{% endif %}