Add pipe support to client, multi-broadcast draft

ffmpstream-client.py

@@ -21,25 +21,42 @@ sp = pUp(s)
 import subprocess
 import socket
 
-bufferSize = 1000
+bufferSize = 1000 # buffer size in bytes
 serverAddr = ("127.0.0.1",12000)
 connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
+def listToCommand(lst):
+	cmd = ""
+	for arg in lst:
+		arg = arg.replace("\\","\\\\")
+		arg = arg.replace(",","\\,")
+		#arg = arg.replace('"','\\"')
+		#if " " in arg: arg = '"' +arg+ '"'
+		cmd += arg + ","
+	
+	return cmd[:-1]
+
+def makePayload(lst):
+	cmdText = listToCommand(lst)
+	cmdBytes = cmdText.encode("utf-8")
+	while len(cmdBytes) < 1000:
+		cmdBytes += b" "
+	return cmdBytes
+
 def main():
 	connection.settimeout(15)
 	connection.connect(serverAddr)
+	connection.sendall(makePayload(sys.argv[1:]))
 	
-	proc = subprocess.Popen([
-		"ffplay","-f","mpegts",
-		"-i","-",
-		"-fflags","nobuffer",
-		"-flags","low_delay",
-		"-infbuf","-fast","-framedrop"
-	],stdin=subprocess.PIPE)
+	if sys.argv[1] == "watch":
+		while True:
+			data = connection.recv(bufferSize)
+			sys.stdout.buffer.write(data)
 	
-	while True:
-		data = connection.recv(bufferSize)
-		proc.stdin.write(data)
+	if sys.argv[1] == "broadcast":
+		while True:
+			data = sys.stdin.buffer.read(bufferSize)
+			connection.sendall(data)
 
 if __name__ == '__main__':
 	main()

ffmpstream-server.py

@@ -23,7 +23,8 @@ import queue
 import socket
 import subprocess
 
-threads = {}
+outThreads = {}
+inThreads = {}
 threadId = 0
 threadsLock = threading.Lock()
 
@@ -32,70 +33,118 @@ serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
 bufferSize = 1000 # buffer size in bytes
 
-proc = subprocess.Popen([
-	# binary
-	"ffmpeg",
-	# input: audio
-	"-f","dshow","-audio_buffer_size","10","-i","audio=virtual-audio-capturer",
-	# input: video
-	"-f","gdigrab","-framerate","30","-i","desktop","-vf","scale=-1:480",
-	# output-codec: video
-	"-c:v","libx264","-pix_fmt","yuv420p","-preset","fast","-tune","zerolatency",
-	# output-codec: audio
-	"-c:a","aac",
-	# output properties:
-	"-bufsize","2M","-maxrate","1M",
-	# output-file:
-	"-f","mpegts",
-	"-"
-],stdout=subprocess.PIPE,stdin=subprocess.PIPE)
+def commandToList(cmd):
+	args = []
+	cArg = ""
+	escape = False
+	quoted = False
+	for letter in cmd:
+		if escape == True:
+			cArg += letter
+			escape = False
+			continue
+		
+		if letter == "\\":
+			escape = True
+			continue
+		
+		#if quoted == False and letter == ",":
+		if letter == ",":
+			if cArg == "": continue
+			args.append(cArg)
+			cArg = ""
+			continue
+		
+		#if letter == '"':
+		#	quoted = not quoted
+		#	continue
+		
+		cArg += letter
+	
+	args.append(cArg)
+	
+	return args
 
 class outThread(threading.Thread):
-	def __init__(self,threadId,connection,address):
+	def __init__(self,threadId,connection,address,user):
 		threading.Thread.__init__(self)
 		self.threadId = threadId
 		self.queue = queue.Queue()
 		self.connection = connection
 		self.address = address
-		self.connection.settimeout(15)
+		self.user = user
+	
+	def closeThread(self):
+		with threadsLock:
+			del outThreads[str(self.threadId)]
 	
 	def run(self):
 		try:
 			while True:
 				data = self.queue.get()
 				self.connection.sendall(data)
+			self.closeThread()
 		except:
-			with threadsLock:
-				del threads[str(self.threadId)]
+			self.closeThread()
 			raise
 
-class ffmpegThread(threading.Thread):
-	def __init__(self):
+class inThread(threading.Thread):
+	def __init__(self,threadId,connection,address):
 		threading.Thread.__init__(self)
+		self.threadId = threadId
+		self.connection = connection
+		self.address = address
+	
+	def closeThread(self):
+		with threadsLock:
+			del inThreads[str(self.threadId)]
 	
 	def run(self):
-		while proc.poll() is None:
-			data = proc.stdout.read(bufferSize)
-			with threadsLock:
-				for thread in threads:
-					thread = threads[thread]
-					thread.queue.put(data)
+		try:
+			global threadId
+			data = self.connection.recv(1000).decode("utf-8")
+			while data[-1] == " ": data = data[:-1]
+			args = commandToList(data)
+			cmd = args.pop(0)
+			
+			user = args[0]
+			self.user = user
+			if len(args) > 1: self.password = args[1]
+			
+			if cmd == "watch":
+				with threadsLock:
+					thread = outThread(self.threadId,self.connection,self.address,self.user)
+					outThreads[str(threadId)] = thread
+					thread.start()
+				return
+			
+			if cmd == "broadcast":
+				while True:
+					data = self.connection.recv(bufferSize)
+					with threadsLock:
+						for thread in outThreads:
+							thread = outThreads[thread]
+							if thread.user == user:
+								thread.queue.put(data)
+			
+			self.closeThread()
+		except:
+			self.closeThread()
+			raise
 
 def main():
 	global threadId
 	serverSocket.bind(serverAddr)
 	serverSocket.listen(1024)
-	ffmpegThread().start()
 	
 	while True:
 		connection, address = serverSocket.accept()
+		connection.settimeout(15)
 		with threadsLock:
 			threadId += 1
-			while str(threadId) in threads:
-				threadId += 1
 			
-			thread = outThread(threadId,connection,address)
-			threads[str(threadId)] = thread
+			thread = inThread(threadId,connection,address)
+			inThreads[str(threadId)] = thread
 			thread.start()
 
 if __name__ == '__main__':