<commit_before>import sys
import socket
import threading
import time
from optparse import OptionParser
import os
import Queue
import re

class StreamingClient:

	def __init__(self, sock):
		self.sock = sock
		sock.settimeout(5) #long timeout to allow clients some flexibility
		self.streamBuffer = ""
		self.streamQueue = Queue.Queue()
		self.streamThread = threading.Thread(target = self.stream)
		self.streamThread.daemon = True
		self.connected = True
		self.kill = False

	def start(self):
		self.streamThread.start()

	def bufferStreamData(self, data):
		#use a thread-safe queue to ensure stream buffer is not modified while we're sending it
		self.streamQueue.put(data)

	def stream(self):
		while True:
			if (self.kill):
				self.sock.close()
				return

			if (not self.streamQueue.empty()):
				self.streamBuffer += self.streamQueue.get()

			if (len(self.streamBuffer) > 0):
				try:
					streamedTo = self.sock.send(self.streamBuffer)
					self.streamBuffer = self.streamBuffer[streamedTo:]
				except socket.error, e:
					self.connected = False
					self.sock.close()
					return

class RequestHandler:
	"""Handles the initial connection with the client"""

	def __init__(self, port, broadcast, status):
		#response we'll send to the client, pretending to be from the real stream source
		dummyHeaderfh = open('dummy.header', 'r')
		self.dummyHeader = dummyHeaderfh.read()

		cssfh = open('style.css', 'r')
		self.statusCSS = cssfh.read()

		htmlfh = open('status.html', 'r')
		self.statusHTML = htmlfh.read()

		self.acceptsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self.acceptsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		self.acceptsock.bind(("0.0.0.0", port))
		self.acceptsock.listen(10)

		self.broadcast = broadcast
		self.status = status

		self.kill = False

		self.acceptThread = threading.Thread(target = self.acceptClients)
		self.acceptThread.daemon = True

	def start(self):
		self.acceptThread.start()

	#
	# Thread to process client requests
	#
	def handleRequest(self, clientsock):
		buff = ""
		while True:
			try:
				data = clientsock.recv(64)
				if (data == ""):
					break

				buff += data

				if "\r\n\r\n" in buff or "\n\n" in buff:
					break #as soon as the header is sent - we only care about GET requests

			except Exception, e:
				print e
				break
		
		if (buff != ""):
			try:
				match = re.search(r'GET (.*) ', buff)

				requestPath = match.group(1)
			except Exception, e:
				print "Client sent unexpected request: {}".format(buff)
				return

			#explicitly deal with individual requests. Verbose, but more secure
			if ("/status" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\nContentType: text/html\r\n\r\n')
				clientsock.sendall(self.statusHTML.format(clientcount = self.broadcast.clientCount, bwin = float(self.status.bandwidthIn*8)/1000000, bwout = float(self.status.bandwidthOut*8)/1000000))
				clientsock.close()
			elif ("/style.css" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\nContentType: text/html\r\n\r\n')
				clientsock.sendall(self.statusCSS)
				clientsock.close()
			elif ("/stream" in requestPath):
				print "Client connected, sending dummy header"
				clientsock.sendall(self.dummyHeader.format(boundaryKey = self.broadcast.boundarySeparator))
				client = StreamingClient(clientsock)
				client.start()
				print "Adding client to join waiting queue"
				self.broadcast.joiningClients.put(client) #blocking, no timeout
			elif ("/snapshot" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\n')
				clientsock.sendall(self.broadcast.lastFrame)
				clientsock.close()
			else:
				clientsock.close()
		else:
			print "Client connected but didn't make a request"

	#
	# Thread to handle connecting clients
	#
	def acceptClients(self):
		while True:
			clientsock, addr = self.acceptsock.accept()

			if (self.kill == True):
				clientsock.close()
				return
			handlethread = threading.Thread(target = self.handleRequest, args = (clientsock,))
			handlethread.start()

class Broadcaster:
	"""Handles relaying the source MJPEG stream to connected clients"""

	def __init__(self, address, port, url, status):
		self.headerType = "multipart/x-mixed-replace"

		self.clients = []
		self.joiningClients = Queue.Queue()
		self.clientCount = 0

		self.status = status

		self.kill = False
		self.broadcastThread = threading.Thread(target = self.streamFromSource)
		self.broadcastThread.daemon = True

		try:
			self.sourcesock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
			self.sourcesock.connect((address, port))
			print "Connected to stream source"
		except Exception, e:
			print "Error: Unable to connect to stream source at {}:{}: {}".format(address, port, e)
			return

		self.boundarySeparator = self.parseStreamHeader(self.getStreamHeader(self.sourcesock, url))

		if (not self.boundarySeparator):
			print "Unable to find boundary separator in the header returned from the stream source"
			return

		print "Boundary separator: {}".format(self.boundarySeparator)

		self.lastFrame = ""
		self.lastFrameBuffer = ""

	def start(self):
		self.broadcastThread.start()<commit_msg>Fix Broadcaster class to prevent broadcast thread being started if stream source is not connected<commit_after>import sys
import socket
import threading
import time
from optparse import OptionParser
import os
import Queue
import re

class StreamingClient:

	def __init__(self, sock):
		self.sock = sock
		sock.settimeout(5) #long timeout to allow clients some flexibility
		self.streamBuffer = ""
		self.streamQueue = Queue.Queue()
		self.streamThread = threading.Thread(target = self.stream)
		self.streamThread.daemon = True
		self.connected = True
		self.kill = False

	def start(self):
		self.streamThread.start()

	def bufferStreamData(self, data):
		#use a thread-safe queue to ensure stream buffer is not modified while we're sending it
		self.streamQueue.put(data)

	def stream(self):
		while True:
			if (self.kill):
				self.sock.close()
				return

			if (not self.streamQueue.empty()):
				self.streamBuffer += self.streamQueue.get()

			if (len(self.streamBuffer) > 0):
				try:
					streamedTo = self.sock.send(self.streamBuffer)
					self.streamBuffer = self.streamBuffer[streamedTo:]
				except socket.error, e:
					self.connected = False
					self.sock.close()
					return

class RequestHandler:
	"""Handles the initial connection with the client"""

	def __init__(self, port, broadcast, status):
		#response we'll send to the client, pretending to be from the real stream source
		dummyHeaderfh = open('dummy.header', 'r')
		self.dummyHeader = dummyHeaderfh.read()

		cssfh = open('style.css', 'r')
		self.statusCSS = cssfh.read()

		htmlfh = open('status.html', 'r')
		self.statusHTML = htmlfh.read()

		self.acceptsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self.acceptsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		self.acceptsock.bind(("0.0.0.0", port))
		self.acceptsock.listen(10)

		self.broadcast = broadcast
		self.status = status

		self.kill = False

		self.acceptThread = threading.Thread(target = self.acceptClients)
		self.acceptThread.daemon = True

	def start(self):
		self.acceptThread.start()

	#
	# Thread to process client requests
	#
	def handleRequest(self, clientsock):
		buff = ""
		while True:
			try:
				data = clientsock.recv(64)
				if (data == ""):
					break

				buff += data

				if "\r\n\r\n" in buff or "\n\n" in buff:
					break #as soon as the header is sent - we only care about GET requests

			except Exception, e:
				print e
				break
		
		if (buff != ""):
			try:
				match = re.search(r'GET (.*) ', buff)

				requestPath = match.group(1)
			except Exception, e:
				print "Client sent unexpected request: {}".format(buff)
				return

			#explicitly deal with individual requests. Verbose, but more secure
			if ("/status" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\nContentType: text/html\r\n\r\n')
				clientsock.sendall(self.statusHTML.format(clientcount = self.broadcast.clientCount, bwin = float(self.status.bandwidthIn*8)/1000000, bwout = float(self.status.bandwidthOut*8)/1000000))
				clientsock.close()
			elif ("/style.css" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\nContentType: text/html\r\n\r\n')
				clientsock.sendall(self.statusCSS)
				clientsock.close()
			elif ("/stream" in requestPath):
				print "Client connected, sending dummy header"
				clientsock.sendall(self.dummyHeader.format(boundaryKey = self.broadcast.boundarySeparator))
				client = StreamingClient(clientsock)
				client.start()
				print "Adding client to join waiting queue"
				self.broadcast.joiningClients.put(client) #blocking, no timeout
			elif ("/snapshot" in requestPath):
				clientsock.sendall('HTTP/1.0 200 OK\r\n')
				clientsock.sendall(self.broadcast.lastFrame)
				clientsock.close()
			else:
				clientsock.close()
		else:
			print "Client connected but didn't make a request"

	#
	# Thread to handle connecting clients
	#
	def acceptClients(self):
		while True:
			clientsock, addr = self.acceptsock.accept()

			if (self.kill == True):
				clientsock.close()
				return
			handlethread = threading.Thread(target = self.handleRequest, args = (clientsock,))
			handlethread.start()

class Broadcaster:
	"""Handles relaying the source MJPEG stream to connected clients"""

	def __init__(self, address, port, url, status):
		self.headerType = "multipart/x-mixed-replace"
		self.url = url

		self.clients = []
		self.joiningClients = Queue.Queue()
		self.clientCount = 0

		self.status = status

		self.kill = False
		self.broadcastThread = threading.Thread(target = self.streamFromSource)
		self.broadcastThread.daemon = True

		self.lastFrame = ""
		self.lastFrameBuffer = ""

	def start(self):
		if (self.connectToStream()):
			print "Connected to stream source, boundary separator: {}".format(self.boundarySeparator)
			self.broadcastThread.start()

	#
	# Connects to the stream source
	#
	def connectToStream(self):
		try:
			self.sourcesock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
			self.sourcesock.connect((address, port))
		except Exception, e:
			print "Error: Unable to connect to stream source at {}:{}: {}".format(address, port, e)
			return False

		self.boundarySeparator = self.parseStreamHeader(self.getStreamHeader(self.sourcesock, self.url))

		if (not self.boundarySeparator):
			print "Unable to find boundary separator in the header returned from the stream source"
			return False

		return True
{'fraction_non_alphanumeric': 0.11591992373689228, 'fraction_numerical': 0.008388941849380362, 'mean_word_length': 3.8013729977116704, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3412054', 'n_tokens_mistral': 3931, 'n_tokens_neox': 3310, 'n_words': 865}