<commit_before># *****************************************************************************
# Copyright (c) 2014, 2017 IBM Corporation and other Contributors.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v1.0
# which accompanies this distribution, and is available at
# http://www.eclipse.org/legal/epl-v10.html
#
# Contributors:
#   <NAME>
#   <NAME>
#   <NAME>
#   <NAME>
#   <NAME>
# *****************************************************************************

import sys
import os
import time
import json
import socket
import ssl
import logging
from logging.handlers import RotatingFileHandler
import paho.mqtt.client as paho
import threading
import iso8601
import pytz
from datetime import datetime
from encodings.base64_codec import base64_encode

__version__ = "0.2.8"

class Message:
	def __init__(self, data, timestamp=None):
		self.data = data
		self.timestamp = timestamp

class AbstractClient:
	def __init__(self, domain, organization, clientId, username, password, port=8883,
										logHandlers=None, cleanSession="true"):
		self.organization = organization
		self.username = username
		self.password = password
		self.address = organization + ".messaging." + domain
		self.port = port
		self.keepAlive = 60

		self.connectEvent = threading.Event()

		self._recvLock = threading.Lock()
		self._messagesLock = threading.Lock()

		self.messages = 0
		self.recv = 0

		self.clientId = clientId

		# Configure logging
		self.logger = logging.getLogger(self.__module__+"."+self.__class__.__name__)
		self.logger.setLevel(logging.INFO)

		# Remove any existing log handlers we may have picked up from getLogger()
		self.logger.handlers = []

		if logHandlers:
			if isinstance(logHandlers, list):
				# Add all supplied log handlers
				for handler in logHandlers:
					self.logger.addHandler(handler)
			else:
				# Add the supplied log handler
				self.logger.addHandler(logHandlers)
		else:
			# Generate a default rotating file log handler and stream handler
			logFileName = '%s.log' % (clientId.replace(":", "_"))
			fhFormatter = logging.Formatter('%(asctime)-25s %(name)-25s ' + ' %(levelname)-7s %(message)s')
			rfh = RotatingFileHandler(logFileName, mode='a', maxBytes=1024000 , backupCount=0, encoding=None, delay=True)
			rfh.setFormatter(fhFormatter)

			ch = logging.StreamHandler()
			ch.setFormatter(fhFormatter)
			ch.setLevel(logging.DEBUG)

			self.logger.addHandler(rfh)
			self.logger.addHandler(ch)

		self.client = paho.Client(self.clientId, clean_session=False if cleanSession == "false" else True)

		try:
			self.tlsVersion = ssl.PROTOCOL_TLSv1_2
		except:
			self.tlsVersion = None

		# Configure authentication
		if self.username is not None:
			# In environments where either ssl is not available, or TLSv1.2 is not available we will fallback to MQTT over TCP
			if self.tlsVersion is not None:
				# Path to certificate
				caFile = os.path.dirname(os.path.abspath(__file__)) + "/messaging.pem"
				self.client.tls_set(ca_certs=caFile, certfile=None, keyfile=None, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2)
			else:
				self.port = 1883
				self.logger.warning("Unable to encrypt messages because TLSv1.2 is unavailable (MQTT over SSL requires at least Python v2.7.9 or 3.4 and openssl v1.0.1)")
			self.client.username_pw_set(self.username, self.password)

		# Attach MQTT callbacks
		self.client.on_log = self.on_log
		self.client.on_connect = self.on_connect
		self.client.on_disconnect = self.on_disconnect
		self.client.on_publish = self.on_publish

		# Initialize default message encoders and decoders.
		self._messageEncoderModules = {}

		self.start = time.time()

		# initialize callbacks
		self._onPublishCallbacks = {}


	def getMessageEncoderModule(self, messageFormat):
		return self._messageEncoderModules[messageFormat]

	def setMessageEncoderModule(self, messageFormat, module):
		self._messageEncoderModules[messageFormat] = module

	def logAndRaiseException(self, e):
		self.logger.critical(str(e))
		raise e

	def connect(self):
		self.logger.debug("Connecting... (address = %s, port = %s, clientId = %s, username = %s, password = %s)" % (self.address, self.port, self.clientId, self.username, self.password))
		try:
			self.connectEvent.clear()
			self.client.connect(self.address, port=self.port, keepalive=self.keepAlive)
			self.client.loop_start()
			if not self.connectEvent.wait(timeout=10):
				self.logAndRaiseException(ConnectionException("Operation timed out connecting to the IBM Internet of Things service: %s" % (self.address)))

		except socket.error as serr:
			self.client.loop_stop()
			self.logAndRaiseException(ConnectionException("Failed to connect to the IBM Internet of Things service: %s - %s" % (self.address, str(serr))))

	def disconnect(self):
		#self.logger.info("Closing connection to the IBM Watson IoT Platform")
		self.client.disconnect()
		# If we don't call loop_stop() it appears we end up with a zombie thread which continues to process
		# network traffic, preventing any subsequent attempt to reconnect using connect()
		self.client.loop_stop()
		#self.stats()
		self.logger.info("Closed connection to the IBM Watson IoT Platform")

	def stats(self):
		elapsed = ((time.time()) - self.start)

		msgPerSecond = 0 if self.messages == 0 else elapsed/self.messages
		recvPerSecond = 0 if self.recv == 0 else elapsed/self.recv
		self.logger.debug("Messages published : %s, life: %.0fs, rate: 1/%.2fs" % (self.messages, elapsed, msgPerSecond))
		self.logger.debug("Messages received  : %s, life: %.0fs, rate: 1/%.2fs" % (self.recv, elapsed, recvPerSecond))
<commit_msg>Stop mqtt thread on connect timeout<commit_after># *****************************************************************************
# Copyright (c) 2014, 2017 IBM Corporation and other Contributors.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v1.0
# which accompanies this distribution, and is available at
# http://www.eclipse.org/legal/epl-v10.html
#
# Contributors:
#   <NAME>
#   <NAME>
#   <NAME>
#   <NAME>
#   <NAME>
# *****************************************************************************

import sys
import os
import time
import json
import socket
import ssl
import logging
from logging.handlers import RotatingFileHandler
import paho.mqtt.client as paho
import threading
import iso8601
import pytz
from datetime import datetime
from encodings.base64_codec import base64_encode

__version__ = "0.2.8"

class Message:
	def __init__(self, data, timestamp=None):
		self.data = data
		self.timestamp = timestamp

class AbstractClient:
	def __init__(self, domain, organization, clientId, username, password, port=8883,
										logHandlers=None, cleanSession="true"):
		self.organization = organization
		self.username = username
		self.password = password
		self.address = organization + ".messaging." + domain
		self.port = port
		self.keepAlive = 60

		self.connectEvent = threading.Event()

		self._recvLock = threading.Lock()
		self._messagesLock = threading.Lock()

		self.messages = 0
		self.recv = 0

		self.clientId = clientId

		# Configure logging
		self.logger = logging.getLogger(self.__module__+"."+self.__class__.__name__)
		self.logger.setLevel(logging.INFO)

		# Remove any existing log handlers we may have picked up from getLogger()
		self.logger.handlers = []

		if logHandlers:
			if isinstance(logHandlers, list):
				# Add all supplied log handlers
				for handler in logHandlers:
					self.logger.addHandler(handler)
			else:
				# Add the supplied log handler
				self.logger.addHandler(logHandlers)
		else:
			# Generate a default rotating file log handler and stream handler
			logFileName = '%s.log' % (clientId.replace(":", "_"))
			fhFormatter = logging.Formatter('%(asctime)-25s %(name)-25s ' + ' %(levelname)-7s %(message)s')
			rfh = RotatingFileHandler(logFileName, mode='a', maxBytes=1024000 , backupCount=0, encoding=None, delay=True)
			rfh.setFormatter(fhFormatter)

			ch = logging.StreamHandler()
			ch.setFormatter(fhFormatter)
			ch.setLevel(logging.DEBUG)

			self.logger.addHandler(rfh)
			self.logger.addHandler(ch)

		self.client = paho.Client(self.clientId, clean_session=False if cleanSession == "false" else True)

		try:
			self.tlsVersion = ssl.PROTOCOL_TLSv1_2
		except:
			self.tlsVersion = None

		# Configure authentication
		if self.username is not None:
			# In environments where either ssl is not available, or TLSv1.2 is not available we will fallback to MQTT over TCP
			if self.tlsVersion is not None:
				# Path to certificate
				caFile = os.path.dirname(os.path.abspath(__file__)) + "/messaging.pem"
				self.client.tls_set(ca_certs=caFile, certfile=None, keyfile=None, cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2)
			else:
				self.port = 1883
				self.logger.warning("Unable to encrypt messages because TLSv1.2 is unavailable (MQTT over SSL requires at least Python v2.7.9 or 3.4 and openssl v1.0.1)")
			self.client.username_pw_set(self.username, self.password)

		# Attach MQTT callbacks
		self.client.on_log = self.on_log
		self.client.on_connect = self.on_connect
		self.client.on_disconnect = self.on_disconnect
		self.client.on_publish = self.on_publish

		# Initialize default message encoders and decoders.
		self._messageEncoderModules = {}

		self.start = time.time()

		# initialize callbacks
		self._onPublishCallbacks = {}


	def getMessageEncoderModule(self, messageFormat):
		return self._messageEncoderModules[messageFormat]

	def setMessageEncoderModule(self, messageFormat, module):
		self._messageEncoderModules[messageFormat] = module

	def logAndRaiseException(self, e):
		self.logger.critical(str(e))
		raise e

	def connect(self):
		self.logger.debug("Connecting... (address = %s, port = %s, clientId = %s, username = %s, password = %s)" % (self.address, self.port, self.clientId, self.username, self.password))
		try:
			self.connectEvent.clear()
			self.client.connect(self.address, port=self.port, keepalive=self.keepAlive)
			self.client.loop_start()
			if not self.connectEvent.wait(timeout=10):
				self.client.loop_stop()
				self.logAndRaiseException(ConnectionException("Operation timed out connecting to the IBM Internet of Things service: %s" % (self.address)))

		except socket.error as serr:
			self.client.loop_stop()
			self.logAndRaiseException(ConnectionException("Failed to connect to the IBM Internet of Things service: %s - %s" % (self.address, str(serr))))

	def disconnect(self):
		#self.logger.info("Closing connection to the IBM Watson IoT Platform")
		self.client.disconnect()
		# If we don't call loop_stop() it appears we end up with a zombie thread which continues to process
		# network traffic, preventing any subsequent attempt to reconnect using connect()
		self.client.loop_stop()
		#self.stats()
		self.logger.info("Closed connection to the IBM Watson IoT Platform")

	def stats(self):
		elapsed = ((time.time()) - self.start)

		msgPerSecond = 0 if self.messages == 0 else elapsed/self.messages
		recvPerSecond = 0 if self.recv == 0 else elapsed/self.recv
		self.logger.debug("Messages published : %s, life: %.0fs, rate: 1/%.2fs" % (self.messages, elapsed, msgPerSecond))
		self.logger.debug("Messages received  : %s, life: %.0fs, rate: 1/%.2fs" % (self.recv, elapsed, recvPerSecond))

{'fraction_non_alphanumeric': 0.13166199158485273, 'fraction_numerical': 0.01332398316970547, 'mean_word_length': 4.487734487734488, 'pattern_counts': {'":': 2, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4590879', 'n_tokens_mistral': 3937, 'n_tokens_neox': 3457, 'n_words': 956}