<commit_before>import asyncore
import Queue
import json
import re
import time
import socket

class DARNMessage:
	def __init__(self, type, expire):
		self.type = type
		# XXX: expire naleven
		self.expire = expire

	def toString(self):
		return json.dumps([self.type, self.getPayload()])

	def getPayload(self):
		return None

class DARNMessagePing(DARNMessage):
	def __init__(self, configversion, expiry):
		DARNMessage.__init__(self, "ping", expiry)
		self.configversion = configversion

	def getPayload(self):
		return [self.configversion]

class DARNMessagePong(DARNMessage):
	def __init__(self, expiry):
		DARNMessage.__init__(self, "pong", expiry)

class DARNHost:
	def __init__(self, connect_callback, data_callback):
		self.host = None
		self.port = None
		self.socket = None
		self.msgqueue = Queue.Queue(0)
		self.connect_callback = connect_callback
		self.data_callback = data_callback

	def setSocket(self, sock):
		self.socket = sock

	def setHost(self, host, port):
		self.host = host
		self.port = port

	def change_callbacks(self, connect_callback, data_callback):
		self.connect_callback = connect_callback
		self.data_callback = data_callback

	def connect(self):
		sock = DARNSocket(self)
		sock.connect(self.host, self.port)
		self.setSocket(sock)

	def has_socket(self):
		return (self.socket is not None)

	def handle_connect(self):
		self.connect_callback(self)

	def receive_msg(self, msg):
		data = json.loads(msg)
		self.data_callback(self, data)

	def send(self, message):
		self.msgqueue.put_nowait(message)
		if not self.has_socket():
			self.connect()

	def send_priority(self, message):
		newq = Queue.Queue(0)
		newq.put_nowait(message)
		while not self.msgqueue.empty():
			newq.put_nowait(self.msgqueue.get_nowait())
		self.msgqueue = newq
		if not self.has_socket():
			self.connect()

	def merge(self, other):
		assert self.host == other.host
		assert self.port == other.port
		other.setSocket(self.socket)
		while not self.msgqueue.empty():
			other.msgqueue.put_nowait(self.msgqueue.get_nowait())
		self.destroy()

	def lost_socket(self):
		self.socket = None
		if not self.host:
			self.destroy()

	def destroy(self):
		self.callback = None
		self.msgqueue = None
		if self.has_socket():
			self.socket.close()
		self.socket = None

class DARNSocket(asyncore.dispatcher):
	def __init__(self, manager, *args):
		asyncore.dispatcher.__init__(self, *args)
		self.manager = manager
		self.outbuf = ''
		self.inbuf = ''

	def connect(self, host, port):
		self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
		asyncore.dispatcher.connect(self, (host, port))

	def handle_connect(self):
		self.manager.handle_connect()

	def handle_close(self):
		self.manager.lost_socket()
		self.close()

	def handle_read(self):
		self.inbuf += self.recv(8192)
		# XXX: re precompilen
		# XXX: connectie sluiten bij invalid data
		m = re.match(r"^(\d+):", self.inbuf)
		if m:
			datalen = len(m.group(0)) + int(m.group(1)) + 1
			if len(self.inbuf) >= datalen:
				self.manager.receive_msg(self.inbuf[len(m.group(0)):datalen-1])
				self.inbuf = self.inbuf[datalen:]
		elif not re.match(r"^\d*", self.inbuf):
			self.close()

	def writable(self):
		if len(self.outbuf) > 0:
			return True
		return (not self.manager.msgqueue.empty())

	def handle_write(self):
		if len(self.outbuf) == 0:
			msg = self.manager.msgqueue.get_nowait()
			str = json.dumps(msg)
			self.outbuf = "%d:%s\n" % (len(str), str)
		sent = self.send(self.outbuf)<commit_msg>Fix merging sockets and fix reading two messages at the same time from a socket<commit_after>import asyncore
import Queue
import json
import re
import time
import socket

class DARNMessage:
	def __init__(self, type, expire):
		self.type = type
		# XXX: expire naleven
		self.expire = expire

	def toString(self):
		return json.dumps([self.type, self.getPayload()])

	def getPayload(self):
		return None

class DARNMessagePing(DARNMessage):
	def __init__(self, configversion, expiry):
		DARNMessage.__init__(self, "ping", expiry)
		self.configversion = configversion

	def getPayload(self):
		return [self.configversion]

class DARNMessagePong(DARNMessage):
	def __init__(self, expiry):
		DARNMessage.__init__(self, "pong", expiry)

class DARNHost:
	def __init__(self, connect_callback, data_callback):
		self.host = None
		self.port = None
		self.socket = None
		self.msgqueue = Queue.Queue(0)
		self.connect_callback = connect_callback
		self.data_callback = data_callback

	def setSocket(self, sock):
		self.socket = sock

	def setHost(self, host, port):
		self.host = host
		self.port = port

	def change_callbacks(self, connect_callback, data_callback):
		self.connect_callback = connect_callback
		self.data_callback = data_callback

	def connect(self):
		sock = DARNSocket(self)
		sock.connect(self.host, self.port)
		self.setSocket(sock)

	def has_socket(self):
		return (self.socket is not None)

	def handle_connect(self):
		self.connect_callback(self)

	def receive_msg(self, msg):
		data = json.loads(msg)
		self.data_callback(self, data)

	def send(self, message):
		self.msgqueue.put_nowait(message)
		if not self.has_socket():
			self.connect()

	def send_priority(self, message):
		newq = Queue.Queue(0)
		newq.put_nowait(message)
		while not self.msgqueue.empty():
			newq.put_nowait(self.msgqueue.get_nowait())
		self.msgqueue = newq
		if not self.has_socket():
			self.connect()

	def merge(self, other):
		assert self.host == other.host
		assert self.port == other.port
		self.socket.manager = other
		other.setSocket(self.socket)
		self.socket = None
		while not self.msgqueue.empty():
			other.msgqueue.put_nowait(self.msgqueue.get_nowait())
		self.destroy()

	def lost_socket(self):
		self.socket = None
		if not self.host:
			self.destroy()

	def destroy(self):
		self.connect_callback = None
		self.data_callback = None
		self.msgqueue = None
		if self.has_socket():
			self.socket.close()
		self.socket = None

class DARNSocket(asyncore.dispatcher):
	def __init__(self, manager, *args):
		asyncore.dispatcher.__init__(self, *args)
		self.manager = manager
		self.outbuf = ''
		self.inbuf = ''

	def connect(self, host, port):
		self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
		asyncore.dispatcher.connect(self, (host, port))

	def handle_connect(self):
		self.manager.handle_connect()

	def handle_close(self):
		self.manager.lost_socket()
		self.close()

	def handle_read(self):
		self.inbuf += self.recv(8192)
		# XXX: re precompilen
		m = re.match(r"^(\d+):", self.inbuf)
		while m:
			datalen = len(m.group(0)) + int(m.group(1)) + 1
			if len(self.inbuf) >= datalen:
				self.manager.receive_msg(self.inbuf[len(m.group(0)):datalen-1])
				self.inbuf = self.inbuf[datalen:]
			else:
				break
			m = re.match(r"^(\d+):", self.inbuf)
		if re.match(r"^\D", self.inbuf):
			self.close()

	def writable(self):
		if len(self.outbuf) > 0:
			return True
		return (not self.manager.msgqueue.empty())

	def handle_write(self):
		if len(self.outbuf) == 0:
			msg = self.manager.msgqueue.get_nowait()
			str = json.dumps(msg)
			self.outbuf = "%d:%s\n" % (len(str), str)
		sent = self.send(self.outbuf)
{'fraction_non_alphanumeric': 0.13071065989847716, 'fraction_numerical': 0.0036661026508742244, 'mean_word_length': 4.289336316181954, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '1592673', 'n_tokens_mistral': 2826, 'n_tokens_neox': 2543, 'n_words': 545}