<commit_before>import websocket
import json
import zmq
import collections
import threading

def assureNotClosed(self, ws, timeout = 1):
	ws.timeout = timeout
	with self.assertRaises(websocket.WebSocketTimeoutException):
		ws.recv()

def assureIsClosed(self, ws):
	with self.assertRaises(websocket.WebSocketConnectionClosedException):
		ws.recv()

def registerGateway(self, ws, gatewayID):
	msg = json.dumps(
		{
			"gateway_id" : gatewayID,
			"ip_address" : "192.168.1.1",
			"message_type" : "gateway_register",
			"version" : "v1.0"
		}
	)

	ws.send(msg)
	msg = json.loads(ws.recv())

	self.assertEqual("gateway_accepted", msg["message_type"])
	assureNotClosed(self, ws)

class ZMQConnection:
	def __init__(self, endpoint):
		self._endpoint = endpoint
		self._ctx = zmq.Context()
		self._data = collections.deque()
		self._lock = threading.Lock()
		self._active = threading.Event()
		self._finish = threading.Event()
		self._stop = False

	def open(self):
		self._sock = self._ctx.socket(zmq.PULL)
		self._sock.setsockopt(zmq.LINGER, 0) # avoid 'Address already in use'
		self._sock.bind(self._endpoint)
		self.start()

	def close(self):
		self.stop()
		self._sock.close()

	def __enter__(self):
		self.open()

	def __exit__(self, type, value, traceback):
		self.close()

	def should_stop(self):
		with self._lock:
			return self._stop

	def stop(self, timeout = 5):
		with self._lock:
			self._stop = True

		self._finish.wait(timeout)

	def push_data(self, data):
		with self._lock:
			self._data.append(data)

		self._active.set()

	def _pop_data(self):
		with self._lock:
			return self._data.popleft()

	def pop_data(self, timeout = 10):
		if timeout == 0:
			return self._pop_data()
		else:
			self.wait_for_data(timeout)
			return self._pop_data()

	def has_data(self):
		with self._lock:
			return len(self._data) > 0

	def wait_for_data(self, timeout = 10):<commit_msg>T: gws: increase the default zmq data timeout to 20 seconds<commit_after>import websocket
import json
import zmq
import collections
import threading

def assureNotClosed(self, ws, timeout = 1):
	ws.timeout = timeout
	with self.assertRaises(websocket.WebSocketTimeoutException):
		ws.recv()

def assureIsClosed(self, ws):
	with self.assertRaises(websocket.WebSocketConnectionClosedException):
		ws.recv()

def registerGateway(self, ws, gatewayID):
	msg = json.dumps(
		{
			"gateway_id" : gatewayID,
			"ip_address" : "192.168.1.1",
			"message_type" : "gateway_register",
			"version" : "v1.0"
		}
	)

	ws.send(msg)
	msg = json.loads(ws.recv())

	self.assertEqual("gateway_accepted", msg["message_type"])
	assureNotClosed(self, ws)

class ZMQConnection:
	def __init__(self, endpoint):
		self._endpoint = endpoint
		self._ctx = zmq.Context()
		self._data = collections.deque()
		self._lock = threading.Lock()
		self._active = threading.Event()
		self._finish = threading.Event()
		self._stop = False

	def open(self):
		self._sock = self._ctx.socket(zmq.PULL)
		self._sock.setsockopt(zmq.LINGER, 0) # avoid 'Address already in use'
		self._sock.bind(self._endpoint)
		self.start()

	def close(self):
		self.stop()
		self._sock.close()

	def __enter__(self):
		self.open()

	def __exit__(self, type, value, traceback):
		self.close()

	def should_stop(self):
		with self._lock:
			return self._stop

	def stop(self, timeout = 5):
		with self._lock:
			self._stop = True

		self._finish.wait(timeout)

	def push_data(self, data):
		with self._lock:
			self._data.append(data)

		self._active.set()

	def _pop_data(self):
		with self._lock:
			return self._data.popleft()

	def pop_data(self, timeout = 20):
		if timeout == 0:
			return self._pop_data()
		else:
			self.wait_for_data(timeout)
			return self._pop_data()

	def has_data(self):
		with self._lock:
			return len(self._data) > 0

	def wait_for_data(self, timeout = 20):
{'fraction_non_alphanumeric': 0.13182773109243698, 'fraction_numerical': 0.01050420168067227, 'mean_word_length': 4.154262516914749, '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': '7408735', 'n_tokens_mistral': 1541, 'n_tokens_neox': 1356, 'n_words': 297}