Stream cast (#4727)

* Broadcast Central

* Change name to Stream_Cast

packages/rocketchat-cas/.npm/package/npm-shrinkwrap.json

@@ -86,7 +86,7 @@
     },
     "isarray": {
       "version": "0.0.1",
-      "resolved": "https://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
+      "resolved": "http://registry.npmjs.org/isarray/-/isarray-0.0.1.tgz",
       "from": "isarray@0.0.1"
     },
     "lodash": {

packages/rocketchat-i18n/i18n/en.i18n.json

@@ -1206,6 +1206,8 @@
   "Stats_Total_Users": "Total Users",
   "Status": "Status",
   "Stop_Recording": "Stop Recording",
+  "Stream_Cast_Address": "Stream Cast Address",
+  "Stream_Cast_Address_Description": "IP or Host of your Rocket.Chat central Stream Cast. E.g. `192.168.1.1:3000` or `localhost:4000`",
   "strike": "strike",
   "Subject": "Subject",
   "Submit": "Submit",

packages/rocketchat-lib/server/startup/settings.coffee

@@ -123,6 +123,9 @@ RocketChat.settings.addGroup 'General', ->
 	@section 'Translations', ->
 		@add 'Custom_Translations', '', { type: 'code', public: true }
 
+	@section 'Stream Cast', ->
+		@add 'Stream_Cast_Address', '', { type: 'string' }
+
 RocketChat.settings.addGroup 'Email', ->
 	@section 'Header and Footer', ->
 		@add 'Email_Header', '<table border="0" cellspacing="0" cellpadding="0" width="100%" bgcolor="#f3f3f3" style="color:#4a4a4a;font-family: Helvetica,Arial,sans-serif;font-size:14px;line-height:20px;border-collapse:callapse;border-spacing:0;margin:0 auto"><tr><td style="padding:1em"><table border="0" cellspacing="0" cellpadding="0" align="center" width="100%" style="width:100%;margin:0 auto;max-width:800px"><tr><td bgcolor="#ffffff" style="background-color:#ffffff; border: 1px solid #DDD; font-size: 10pt; font-family: Helvetica,Arial,sans-serif;"><table width="100%" border="0" cellspacing="0" cellpadding="0"><tr><td style="background-color: #04436a;"><h1 style="font-family: Helvetica,Arial,sans-serif; padding: 0 1em; margin: 0; line-height: 70px; color: #FFF;">[Site_Name]</h1></td></tr><tr><td style="padding: 1em; font-size: 10pt; font-family: Helvetica,Arial,sans-serif;">', {

server/stream/streamBroadcast.coffee

+`import {DDPCommon} from 'meteor/ddp-common'`
+
 logger = new Logger 'StreamBroadcast',
 	sections:
 		connection: 'Connection'
@@ -7,9 +9,9 @@ logger = new Logger 'StreamBroadcast',
 _authorizeConnection = (instance) ->
 	logger.auth.info "Authorizing with #{instance}"
 
-	connections[instance].call 'broadcastAuth', connections[instance].instanceRecord._id, InstanceStatus.id(), (err, ok) ->
+	connections[instance].call 'broadcastAuth', InstanceStatus.id(), connections[instance].instanceId, (err, ok) ->
 		if err?
-			return logger.auth.error "broadcastAuth error #{instance} #{connections[instance].instanceRecord._id} #{InstanceStatus.id()}", err
+			return logger.auth.error "broadcastAuth error #{instance} #{InstanceStatus.id()} #{connections[instance].instanceId}", err
 
 		connections[instance].broadcastAuth = ok
 		logger.auth.info "broadcastAuth with #{instance}", ok
@@ -22,12 +24,7 @@ authorizeConnection = (instance) ->
 
 	_authorizeConnection(instance)
 
-@connections = {}
-@startStreamBroadcast = () ->
-	process.env.INSTANCE_IP ?= 'localhost'
-
-	logger.info 'startStreamBroadcast'
-
+startMatrixBroadcast = ->
 	InstanceStatus.getCollection().find({'extraInformation.port': {$exists: true}}, {sort: {_createdAt: -1}}).observe
 		added: (record) ->
 			instance = "#{record.extraInformation.host}:#{record.extraInformation.port}"
@@ -49,6 +46,7 @@ authorizeConnection = (instance) ->
 			logger.connection.info 'connecting in', instance
 			connections[instance] = DDP.connect(instance, {_dontPrintErrors: true})
 			connections[instance].instanceRecord = record;
+			connections[instance].instanceId = record._id;
 			connections[instance].onReconnect = ->
 				authorizeConnection(instance)
 
@@ -63,6 +61,80 @@ authorizeConnection = (instance) ->
 				connections[instance].disconnect()
 				delete connections[instance]
 
+	Meteor.methods
+		broadcastAuth: (remoteId, selfId) ->
+			check selfId, String
+			check remoteId, String
+
+			@unblock()
+			if selfId is InstanceStatus.id() and remoteId isnt InstanceStatus.id() and InstanceStatus.getCollection().findOne({_id: remoteId})?
+				@connection.broadcastAuth = true
+
+			return @connection.broadcastAuth is true
+
+		stream: (streamName, eventName, args) ->
+			# Prevent call from self and client
+			if not @connection?
+				return 'self-not-authorized'
+
+			# Prevent call from unauthrorized connections
+			if @connection.broadcastAuth isnt true
+				return 'not-authorized'
+
+			if not Meteor.StreamerCentral.instances[streamName]?
+				return 'stream-not-exists'
+
+			Meteor.StreamerCentral.instances[streamName]._emit(eventName, args)
+
+			return undefined
+
+startStreamCastBroadcast = (value) ->
+	instance = 'StreamCast'
+
+	logger.connection.info 'connecting in', instance, value
+	connection = DDP.connect(value, {_dontPrintErrors: true})
+	connections[instance] = connection
+	connection.instanceId = instance
+	connection.onReconnect = ->
+		authorizeConnection(instance)
+
+	connection._stream.on 'message', (raw_msg) ->
+		msg = DDPCommon.parseDDP(raw_msg)
+		if not msg or msg.msg isnt 'changed' or not msg.collection? or not msg.fields?
+			return
+
+		{streamName, eventName, args} = msg.fields
+
+		if not streamName? or not eventName? or not args?
+			return
+
+		if connection.broadcastAuth isnt true
+			return 'not-authorized'
+
+		if not Meteor.StreamerCentral.instances[streamName]?
+			return 'stream-not-exists'
+
+		Meteor.StreamerCentral.instances[streamName]._emit(eventName, args)
+
+	connection.subscribe 'stream'
+
+@connections = {}
+@startStreamBroadcast = () ->
+	process.env.INSTANCE_IP ?= 'localhost'
+
+	logger.info 'startStreamBroadcast'
+
+	RocketChat.settings.get 'Stream_Cast_Address', (key, value) ->
+		for instance, connection of connections
+			do (instance, connection) ->
+				connection.disconnect()
+				delete connections[instance]
+
+		if value?.trim() isnt ''
+			startStreamCastBroadcast(value)
+		else
+			startMatrixBroadcast()
+
 	broadcast = (streamName, eventName, args, userId) ->
 		fromInstance = process.env.INSTANCE_IP + ':' + process.env.PORT
 		for instance, connection of connections
@@ -95,32 +167,5 @@ authorizeConnection = (instance) ->
 	Meteor.StreamerCentral.on 'broadcast', (streamName, eventName, args) ->
 		broadcast streamName, eventName, args
 
-	Meteor.methods
-		broadcastAuth: (selfId, remoteId) ->
-			check selfId, String
-			check remoteId, String
-
-			@unblock()
-			if selfId is InstanceStatus.id() and remoteId isnt InstanceStatus.id() and InstanceStatus.getCollection().findOne({_id: remoteId})?
-				@connection.broadcastAuth = true
-
-			return @connection.broadcastAuth is true
-
-		stream: (streamName, eventName, args) ->
-			# Prevent call from self and client
-			if not @connection?
-				return 'self-not-authorized'
-
-			# Prevent call from unauthrorized connections
-			if @connection.broadcastAuth isnt true
-				return 'not-authorized'
-
-			if not Meteor.StreamerCentral.instances[streamName]?
-				return 'stream-not-exists'
-
-			Meteor.StreamerCentral.instances[streamName]._emit(eventName, args)
-
-			return undefined
-
 Meteor.startup ->
 	startStreamBroadcast()