<commit_before>import sys
try:
	from Queue import Queue
except ImportError:
	from queue import Queue
import tempfile
from datetime import datetime
import time

class MemorySubscriber:
	def __init__(self):
		self.messages = Queue()

	def getNext(self):
		if self.messages.qsize() == 0:
			return None
		else:
			return self.messages.get(block=False, timeout=None)

	def getAll(self):
    		items = []
    		maxItemsToRetreive = self.messages.qsize()
		for numOfItemsRetrieved in range(0, maxItemsToRetreive):
        		try:
            			if numOfItemsRetrieved == maxItemsToRetreive:
                			break
            			items.append(self.messages.get_nowait())
        		except Empty, e:
            			break
    		return items

	def getCount(self):
		return self.messages.qsize()

	def put(self, message):
		self.messages.put_nowait(message)

class MemoryPubSub:
	def __init__(self):
		self.subscribers = {}

	def publish(self, topic, message):
		subscribers = self.subscribers.get(topic, [])
		for each_subscriber in subscribers:
			each_subscriber.put(message)
		return "Published to " + str(len(subscribers)) + " subscribers"

	def subscribe(self, *topics):
		subscriber = MemorySubscriber()
		for topic in topics:
			subscribers = self.subscribers.setdefault(topic, [])
			subscribers.append(subscriber)
		return subscriber

class SQLiteSubscriber:
	def __init__(self, cursor, topics, timestamp):
		self.messages = Queue()
		self.timestamp = timestamp
		self.topics = list(topics)
		self.cursor = cursor

	def getNext(self):
		for topic in self.topics:
			self.cursor.execute("SELECT message, timestamp from mps_messages WHERE topic=:topic and timestamp>:timestamp", {"topic": topic, "timestamp": self.timestamp})
			data = self.cursor.fetchall()
			for each_record in data:
				self.messages.put_nowait(each_record[0])
		self.timestamp = datetime.utcnow()
		if self.messages.qsize() == 0:
			return None
		else:
			return self.messages.get(block=False, timeout=None)

	def getAll(self):
		for topic in self.topics:
			self.cursor.execute("SELECT message, timestamp from mps_messages WHERE topic=:topic and timestamp>:timestamp", {"topic": topic, "timestamp": self.timestamp})<commit_msg>Updated description for MemorySubscriber<commit_after>import sys
try:
	from Queue import Queue
except ImportError:
	from queue import Queue
import tempfile
from datetime import datetime
import time

class MemorySubscriber:
	def __init__(self):
		''' Initializes the empty queue for a particular subscriber. '''
		
		self.messages = Queue()

	def getNext(self):
		''' Returns the next message available in the queue. Returns None if queue is empty. '''
		
		if self.messages.qsize() == 0:
			return None
		else:
			return self.messages.get(block=False, timeout=None)

	def getAll(self):
		''' Get all the messages available in the queue, appends them in the 'item' list and returns the list. '''
		
    		items = []
    		maxItemsToRetreive = self.messages.qsize()
		for numOfItemsRetrieved in range(0, maxItemsToRetreive):
        		try:
            			if numOfItemsRetrieved == maxItemsToRetreive:
                			break
            			items.append(self.messages.get_nowait())
        		except Empty, e:
            			break
    		return items

	def getCount(self):
		''' Returns the number of messages available in the queue. '''
		
		return self.messages.qsize()

	def put(self, message):
		''' Puts the message in the queue. '''
		
		self.messages.put_nowait(message)

class MemoryPubSub:
	''' This class is invoked when the user chooses to store the messages queues in the memory '''
	
	def __init__(self):
		''' Intializes with no subscribers available '''
		
		self.subscribers = {}

	def publish(self, topic, message):
		''' When this method is invoked with a topic name and the message, it looks for all the subscribers for that particular topic and pushes the message to the subscribers. Returns the number of subscribers available. '''
		
		subscribers = self.subscribers.get(topic, [])
		for each_subscriber in subscribers:
			each_subscriber.put(message)
		return "Published to " + str(len(subscribers)) + " subscribers"

	def subscribe(self, *topics):
		''' When this method is invoked with a list of topics, it creates a MemorySubscriber() object and for all the topics appends the object into the dictionary and returns the handle to the MemorySubscriber class. '''
		
		subscriber = MemorySubscriber()
		for topic in topics:
			subscribers = self.subscribers.setdefault(topic, [])
			subscribers.append(subscriber)
		return subscriber

class SQLiteSubscriber:
	def __init__(self, cursor, topics, timestamp):
		self.messages = Queue()
		self.timestamp = timestamp
		self.topics = list(topics)
		self.cursor = cursor

	def getNext(self):
		for topic in self.topics:
			self.cursor.execute("SELECT message, timestamp from mps_messages WHERE topic=:topic and timestamp>:timestamp", {"topic": topic, "timestamp": self.timestamp})
			data = self.cursor.fetchall()
			for each_record in data:
				self.messages.put_nowait(each_record[0])
		self.timestamp = datetime.utcnow()
		if self.messages.qsize() == 0:
			return None
		else:
			return self.messages.get(block=False, timeout=None)

	def getAll(self):
		for topic in self.topics:
			self.cursor.execute("SELECT message, timestamp from mps_messages WHERE topic=:topic and timestamp>:timestamp", {"topic": topic, "timestamp": self.timestamp})
{'fraction_non_alphanumeric': 0.09957431056820285, 'fraction_numerical': 0.0014806588932074773, 'mean_word_length': 3.8553459119496853, 'pattern_counts': {'":': 8, '<': 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': '3456636', 'n_tokens_mistral': 1774, 'n_tokens_neox': 1587, 'n_words': 495}