<commit_before>import sys, os, json, requests, threading
from urlparse import urljoin
from pyinotify import WatchManager, Notifier, EventsCodes, ProcessEvent
import converter, settings

class QueueManager:

	def __init__(self):
		self.cache = []
		self.cache_lock = threading.Lock()
		self.publish_timer = None

	def __del__(self):
		self.destroy()
		
	def destroy(self):

		if self.publish_timer != None:
			self.publish_timer.cancel()

	def push(self, stmt):

		with self.cache_lock:
			self.cache.append(stmt)

		# publish if oldest statement reaches timeout
		if len(self.cache) == 1 and settings.PUBLISH_MAX_WAIT_TIME > 0:
			self.publish_timer = threading.Timer(settings.PUBLISH_MAX_WAIT_TIME, self.publish)
			self.publish_timer.start()

		# publish if statement threshold is reached
		if settings.PUBLISH_MAX_PAYLOAD <= len(self.cache):
			self.publish()

	def publish(self):

		with self.cache_lock:
			url = urljoin(settings.LRS_ENDPOINT, 'statements')
			r = requests.post(url, data=json.dumps(self.cache),
				auth=(settings.LRS_USERNAME, settings.LRS_PASSWORD),
				headers={'X-Experience-API-Version':'1.0.1', 'Content-Type':'application/json'})

			print r.text
			self.cache = []
			if self.publish_timer != None:
				self.publish_timer.cancel()


class TailHandler(ProcessEvent):

	MASK = EventsCodes.OP_FLAGS['IN_MODIFY']

	def __init__(self, filename):

		# prepare file input stream
		self.ifp = open(filename, 'r', 1)
		self.ifp.seek(0,2)

		self.publish_queue = QueueManager()

	def __enter__(self):
		return self
	def __exit__(self, type, value, tb):
		self.publish_queue.destroy()
		
	def process_IN_MODIFY(self,event):

		buff = ''
		for line in self.ifp:
			buff += line
			print len(line), '->', len(buff)

		evts = [i for i in buff.split('\n') if len(i) != 0]
		for e in evts:
			try:
				evtObj = json.loads(e)
			except ValueError as err:
				print 'Could not parse JSON for', e
				continue

			xapi = converter.to_xapi(evtObj)
			if xapi != None:
				for i in xapi:<commit_msg>Changed from for line in file to while data read<commit_after>import sys, os, json, requests, threading
from urlparse import urljoin
from pyinotify import WatchManager, Notifier, EventsCodes, ProcessEvent
import converter, settings

class QueueManager:

	def __init__(self):
		self.cache = []
		self.cache_lock = threading.Lock()
		self.publish_timer = None

	def __del__(self):
		self.destroy()
		
	def destroy(self):

		if self.publish_timer != None:
			self.publish_timer.cancel()

	def push(self, stmt):

		with self.cache_lock:
			self.cache.append(stmt)

		# publish if oldest statement reaches timeout
		if len(self.cache) == 1 and settings.PUBLISH_MAX_WAIT_TIME > 0:
			self.publish_timer = threading.Timer(settings.PUBLISH_MAX_WAIT_TIME, self.publish)
			self.publish_timer.start()

		# publish if statement threshold is reached
		if settings.PUBLISH_MAX_PAYLOAD <= len(self.cache):
			self.publish()

	def publish(self):

		with self.cache_lock:
			url = urljoin(settings.LRS_ENDPOINT, 'statements')
			r = requests.post(url, data=json.dumps(self.cache),
				auth=(settings.LRS_USERNAME, settings.LRS_PASSWORD),
				headers={'X-Experience-API-Version':'1.0.1', 'Content-Type':'application/json'})

			print r.text
			self.cache = []
			if self.publish_timer != None:
				self.publish_timer.cancel()


class TailHandler(ProcessEvent):

	MASK = EventsCodes.OP_FLAGS['IN_MODIFY']

	def __init__(self, filename):

		# prepare file input stream
		self.ifp = open(filename, 'r', 1)
		self.ifp.seek(0,2)

		self.publish_queue = QueueManager()

	def __enter__(self):
		return self
	def __exit__(self, type, value, tb):
		self.publish_queue.destroy()
		
	def process_IN_MODIFY(self,event):

		buff = line = self.ifp.read()
		while line != '':
			line = self.ifp.read()
			buff += line
			print len(line), '->', len(buff)

		evts = [i for i in buff.split('\n') if len(i) != 0]
		for e in evts:
			try:
				evtObj = json.loads(e)
			except ValueError as err:
				print 'Could not parse JSON for', e
				continue

			xapi = converter.to_xapi(evtObj)
			if xapi != None:
				for i in xapi:
{'fraction_non_alphanumeric': 0.12234432234432234, 'fraction_numerical': 0.004395604395604396, 'mean_word_length': 3.9290012033694346, 'pattern_counts': {'":': 0, '<': 5, '<?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': '4192460', 'n_tokens_mistral': 1608, 'n_tokens_neox': 1397, 'n_words': 346}