Skip to content

GitLab

Commit 38a24d75 authored by Federico Sismondi's avatar Federico Sismondi
Browse files

plenty of new features to REPL component, the most notorious one is the... 

plenty of new features to REPL component, the most notorious one is the possibility to run it in a lazy mode meaning that messages body are just printed in STDOUT instead of being processes as Messages
parent ca2c1188
Hide whitespace changes
Inline Side-by-side
cli.py
View file @ 38a24d75
...@@ -31,6 +31,8 @@ from pure_pcapy import Dumper, Pkthdr, DLT_IEEE802_15_4, DLT_RAW ...@@ -31,6 +31,8 @@ from pure_pcapy import Dumper, Pkthdr, DLT_IEEE802_15_4, DLT_RAW
# globals # globals
message_count = 0 message_count = 0
print("THIS CLI IS NOT SUPPORTED ANY MORE")
logging.warning("THIS CLI IS NOT SUPPORTED ANY MORE")
def print_message(method, props, body): def print_message(method, props, body):
global message_count global message_count
...@@ -547,29 +549,40 @@ if __name__ == '__main__': ...@@ -547,29 +549,40 @@ if __name__ == '__main__':
}) })
service_tat = OrderedDict({ service_tat = OrderedDict({
'tat0': MsgInteropTestCaseAnalyze(), 'tat0': MsgInteropTestCaseAnalyze(
protocol='coap',
),
'tat1': MsgInteropTestCaseAnalyze( 'tat1': MsgInteropTestCaseAnalyze(
protocol='coap',
testcase_id="TD_COAP_CORE_01", testcase_id="TD_COAP_CORE_01",
testcase_ref="http://f-interop.paris.inria.fr/tests/TD_COAP_CORE_01_v01", testcase_ref="http://doc.f-interop.eu/tests/TD_COAP_CORE_01_v01",
file_enc="pcap_base64", file_enc="pcap_base64",
filename="TD_COAP_CORE_01.pcap", filename="TD_COAP_CORE_01.pcap",
value=PCAP_empty_base64, value=PCAP_empty_base64,
), ),
'tat2': MsgInteropTestCaseAnalyze( 'tat2': MsgInteropTestCaseAnalyze(
protocol='coap',
testcase_id="TD_COAP_CORE_01", testcase_id="TD_COAP_CORE_01",
testcase_ref="http://f-interop.paris.inria.fr/tests/TD_COAP_CORE_01_v01", testcase_ref="http://doc.f-interop.eu/tests/TD_COAP_CORE_01_v01",
file_enc="pcap_base64", file_enc="pcap_base64",
filename="TD_COAP_CORE_01.pcap", filename="TD_COAP_CORE_01.pcap",
value=PCAP_TC_COAP_01_base64, value=PCAP_TC_COAP_01_base64,
), ),
# 'tat3': MsgInteropTestCaseAnalyze( # 'tat3': MsgInteropTestCaseAnalyze(
# testcase_id="TD_COAP_CORE_04", # testcase_id="TD_COAP_CORE_04",
# testcase_ref="http://f-interop.paris.inria.fr/tests/TD_COAP_CORE_04_v01", # testcase_ref="http://doc.f-interop.eu/tests/TD_COAP_CORE_04_v01",
# file_enc="pcap_base64", # file_enc="pcap_base64",
# filename="TD_COAP_CORE_04.pcap", # filename="TD_COAP_CORE_04.pcap",
# value=PCAP_COAP_TC4_OVER_TUN_INTERFACE_base64, # value=PCAP_COAP_TC4_OVER_TUN_INTERFACE_base64,
# ) # )
'tat_onem2m': MsgInteropTestCaseAnalyze(
protocol='onem2m',
testcase_id="TD_M2M_NH_01",
testcase_ref="http://doc.f-interop.eu/tests/TD_M2M_NH_01",
file_enc="pcap_base64",
filename="TD_M2M_NH_01.pcap",
value=PCAP_ONEM2M_TD_M2M_NH_01,
),
}) })
service_dissection = OrderedDict({ service_dissection = OrderedDict({
...@@ -679,16 +692,27 @@ if __name__ == '__main__': ...@@ -679,16 +692,27 @@ if __name__ == '__main__':
'conf_format_04_eut2': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_04', node='eut2'), 'conf_format_04_eut2': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_04', node='eut2'),
'conf_format_06_eut1': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_06', node='eut1'), 'conf_format_06_eut1': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_06', node='eut1'),
'conf_format_06_eut2': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_06', node='eut2'), 'conf_format_06_eut2': MsgConfigurationExecute(testcase_id='TD_6LoWPAN_FORMAT_06', node='eut2'),
's_hc_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_01', node='eut1', target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"), # need to update the target_Address before sending the message!! 's_hc_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_01', node='eut1',
's_hc_03': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_03', node='eut1', target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"),
's_hc_05_step_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_05', node='eut2', target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"), # need to update the target_Address before sending the message!!
's_hc_05_step_02': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_05', node='eut1', target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), 's_hc_03': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_03', node='eut1',
's_hc_07_step_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_07', node='eut2', target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"), target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
's_hc_07_step_02': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_07', node='eut1', target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), 's_hc_05_step_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_05', node='eut2',
's_format_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_01', node='eut1',target_address="fe80:0000:0000:0000:0212:4b00:060d:97f5"), target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"),
's_format_03': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_03', node='eut1',target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), 's_hc_05_step_02': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_05', node='eut1',
's_format_04': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_04', node='eut1',target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
's_format_06': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_06', node='eut1',target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"), 's_hc_07_step_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_07', node='eut2',
target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"),
's_hc_07_step_02': MsgStepStimuliExecute(step_id='TD_6LoWPAN_HC_07', node='eut1',
target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
's_format_01': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_01', node='eut1',
target_address="fe80:0000:0000:0000:0212:4b00:060d:97f5"),
's_format_03': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_03', node='eut1',
target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
's_format_04': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_04', node='eut1',
target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
's_format_06': MsgStepStimuliExecute(step_id='TD_6LoWPAN_FORMAT_06', node='eut1',
target_address="fe80:0000:0000:0000:0212:4b00:0433:ed9c"),
# 'tc_hc_01': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_01'), They have no effect on the iut controller # 'tc_hc_01': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_01'), They have no effect on the iut controller
# 'tc_hc_02': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_02'), # 'tc_hc_02': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_02'),
# 'tc_hc_03': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_03'), # 'tc_hc_03': MsgTestCaseReady(testcase_id='TD_6LoWPAN_HC_03'),
...@@ -713,7 +737,6 @@ if __name__ == '__main__': ...@@ -713,7 +737,6 @@ if __name__ == '__main__':
# target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"), # target_address="fe80:0000:0000:0000:0212:4b00:0615:a500"),
}) })
event_type = params[0] event_type = params[0]
print(event_type) print(event_type)
......
event_bus_utils.py
View file @ 38a24d75
...@@ -24,18 +24,19 @@ class AmqpListener(threading.Thread): ...@@ -24,18 +24,19 @@ class AmqpListener(threading.Thread):
DEFAULT_EXCHAGE = 'amq.topic' DEFAULT_EXCHAGE = 'amq.topic'
DEFAULT_AMQP_URL = 'amqp://guest:guest@locahost/' DEFAULT_AMQP_URL = 'amqp://guest:guest@locahost/'
def __init__(self, amqp_url, amqp_exchange, callback, topics=None): def __init__(self, amqp_url, amqp_exchange, callback, topics=None, use_message_typing=True):
self.COMPONENT_ID = 'amqp_listener_%s' % str(uuid.uuid4())[:8] self.COMPONENT_ID = 'amqp_listener_%s' % str(uuid.uuid4())[:8]
self.connection = None self.connection = None
self.channel = None self.channel = None
self.services_queue_name = 'services_queue@%s' % self.COMPONENT_ID self.services_queue_name = 'services_queue@%s' % self.COMPONENT_ID
self.use_message_typing = use_message_typing
threading.Thread.__init__(self) threading.Thread.__init__(self)
if callback is None: if callback is None:
self.message_dispatcher = print self.message_dispatcher = AmqpListener.default_message_handler
else: else:
self.message_dispatcher = callback self.message_dispatcher = callback
...@@ -56,6 +57,11 @@ class AmqpListener(threading.Thread): ...@@ -56,6 +57,11 @@ class AmqpListener(threading.Thread):
self.amqp_connect() self.amqp_connect()
@classmethod
def default_message_handler(cls,message_as_dict):
clean_dict = dict((k, v) for k, v in message_as_dict.items() if v)
print(json.dumps(clean_dict,indent=4))
def amqp_connect(self): def amqp_connect(self):
self.connection = pika.BlockingConnection(pika.URLParameters(self.amqp_url)) self.connection = pika.BlockingConnection(pika.URLParameters(self.amqp_url))
self.channel = self.connection.channel() self.channel = self.connection.channel()
...@@ -106,25 +112,34 @@ class AmqpListener(threading.Thread): ...@@ -106,25 +112,34 @@ class AmqpListener(threading.Thread):
'app_id': props.app_id, 'app_id': props.app_id,
} }
try: if self.use_message_typing:
m = Message.from_json(body) try:
if m is None: m = Message.from_json(body)
raise Exception("Couldnt build message from json %s, amqp props: %s " % (body, props_dict)) if m is None:
m.update_properties(**props_dict) raise Exception("Couldnt build message from json %s, amqp props: %s " % (body, props_dict))
m.routing_key = method.routing_key m.update_properties(**props_dict)
logging.debug('Message in bus: %s'%repr(m)) m.routing_key = method.routing_key
self.message_dispatcher(m) logging.debug('Message in bus: %s'%repr(m))
self.message_dispatcher(m)
except NonCompliantMessageFormatError as e:
logging.error('%s got a non compliant message error %s' % (self.__class__.__name__, e)) except NonCompliantMessageFormatError as e:
logging.error('%s got a non compliant message error %s' % (self.__class__.__name__, e))
except Exception as e:
logging.error(e) except Exception as e:
logging.error('message received:\n\tr_key: %s\n\t%s' % (method.routing_key, body)) logging.error(e)
raise e logging.error('message received:\n\tr_key: %s\n\t%s' % (method.routing_key, body))
raise e
finally:
finally:
ch.basic_ack(delivery_tag=method.delivery_tag)
else:
body_dict = json.loads(body.decode('utf-8'), object_pairs_hook=OrderedDict)
ch.basic_ack(delivery_tag=method.delivery_tag) ch.basic_ack(delivery_tag=method.delivery_tag)
text_based_message_representation = OrderedDict()
text_based_message_representation.update({'routing_key':method.routing_key})
text_based_message_representation.update(props_dict)
text_based_message_representation.update(body_dict)
self.message_dispatcher(text_based_message_representation)
def run(self): def run(self):
logging.info("Starting thread listening on the event bus on topics %s" % self.topics) logging.info("Starting thread listening on the event bus on topics %s" % self.topics)
...@@ -169,7 +184,7 @@ def publish_message(connection, message): ...@@ -169,7 +184,7 @@ def publish_message(connection, message):
channel.close() channel.close()
def amqp_request(connection, request_message, component_id): def amqp_request(connection, request_message, component_id, retries = 10):
""" """
Publishes message into the correct topic (uses Message object metadata) Publishes message into the correct topic (uses Message object metadata)
Returns reply message. Returns reply message.
...@@ -181,6 +196,9 @@ def amqp_request(connection, request_message, component_id): ...@@ -181,6 +196,9 @@ def amqp_request(connection, request_message, component_id):
# check first that sender didnt forget about reply to and corr id # check first that sender didnt forget about reply to and corr id
assert request_message.reply_to assert request_message.reply_to
assert request_message.correlation_id assert request_message.correlation_id
assert retries > 0
time_between_requests = 0.5
channel = None channel = None
...@@ -207,10 +225,10 @@ def amqp_request(connection, request_message, component_id): ...@@ -207,10 +225,10 @@ def amqp_request(connection, request_message, component_id):
) )
time.sleep(0.2) time.sleep(0.2)
retries_left = 10 retries_left = retries
while retries_left > 0: while retries_left > 0:
time.sleep(0.5) time.sleep(time_between_requests)
method, props, body = channel.basic_get(reply_queue_name) method, props, body = channel.basic_get(reply_queue_name)
if method: if method:
channel.basic_ack(method.delivery_tag) channel.basic_ack(method.delivery_tag)
...@@ -261,37 +279,48 @@ if __name__ == '__main__': ...@@ -261,37 +279,48 @@ if __name__ == '__main__':
print("Callback function received: \n\t" + repr(message_received)) print("Callback function received: \n\t" + repr(message_received))
# amqp listener example: # EXAMPLE ON AMQP LISTENER
amqp_listener_thread = AmqpListener(
amqp_url=AMQP_URL, # # amqp listener example:
amqp_exchange=AMQP_EXCHANGE, # amqp_listener_thread = AmqpListener(
callback=callback_function, # amqp_url=AMQP_URL,
topics='#' # amqp_exchange=AMQP_EXCHANGE,
) # callback=callback_function,
# topics='#'
try: # )
amqp_listener_thread.start() #
except Exception as e: # try:
print(e) # amqp_listener_thread.start()
# except Exception as e:
# publish message example # print(e)
retries_left = 3 #
while retries_left > 0: # # publish message example
try: # retries_left = 3
connection = pika.BlockingConnection(pika.URLParameters(AMQP_URL)) # while retries_left > 0:
m = MsgTest() # try:
publish_message(connection, m) # connection = pika.BlockingConnection(pika.URLParameters(AMQP_URL))
break # m = MsgTest()
except pika.exceptions.ConnectionClosed: # publish_message(connection, m)
retries_left -= 1 # break
print('retrying..') # except pika.exceptions.ConnectionClosed:
time.sleep(0.2) # retries_left -= 1
# print('retrying..')
# example of a request sent into the bus # time.sleep(0.2)
m = MsgTestSuiteGetTestCases() #
try: # # example of a request sent into the bus
r = amqp_request(connection, m, 'someImaginaryComponent') # m = MsgTestSuiteGetTestCases()
print("This is the response I got:\n\t" + repr(r)) # try:
# r = amqp_request(connection, m, 'someImaginaryComponent')
except AmqpSynchCallTimeoutError as e: # print("This is the response I got:\n\t" + repr(r))
print("Nobody answered to our request :'(") #
# except AmqpSynchCallTimeoutError as e:
# print("Nobody answered to our request :'(")
# EXAMPLE ON REQUEST REPLY FOR UI BUTTONS:
con = pika.BlockingConnection(pika.URLParameters(AMQP_URL))
channel = con.channel()
ui_request = MsgUiRequestConfirmationButton()
ui_reply = amqp_request(con,ui_request,'dummy_component')
print(repr(ui_reply))
\ No newline at end of file
interop_cli.py
View file @ 38a24d75
...@@ -47,9 +47,6 @@ DEFAULT_TOPIC_SUBSCRIPTIONS = [ ...@@ -47,9 +47,6 @@ DEFAULT_TOPIC_SUBSCRIPTIONS = [
MESSAGE_TYPES_NOT_ECHOED = [ MESSAGE_TYPES_NOT_ECHOED = [
MsgPacketInjectRaw, MsgPacketInjectRaw,
MsgUiReply,
MsgUiDisplay,
MsgUiDisplayMarkdownText
] ]
CONNECTION_SETUP_RETRIES = 3 CONNECTION_SETUP_RETRIES = 3
...@@ -64,6 +61,16 @@ session_profile = OrderedDict( ...@@ -64,6 +61,16 @@ session_profile = OrderedDict(
} }
) )
# TODO handle the lock automatically with an state object
# def some_dummuy_release():
# _echo_log_message('releasing!')
# def some_dummuy_acquire():
# _echo_log_message('acquiring!')
# state_lock = Message()
# state_lock.__setattr__('release', some_dummuy_release)
# state_lock.__setattr__('acquire', some_dummuy_acquire)
state_lock = threading.RLock()
state = { state = {
'testcase_id': None, 'testcase_id': None,
'step_id': None, 'step_id': None,
...@@ -90,10 +97,12 @@ def _init_action_suggested(): ...@@ -90,10 +97,12 @@ def _init_action_suggested():
state['suggested_cmd'] = 'ts_start' state['suggested_cmd'] = 'ts_start'
def amqp_request(channel, request_message, component_id): def amqp_request(request_message, component_id=COMPONENT_ID):
""" """
NOTE: channel must be a pika channel NOTE: channel must be a pika channel
""" """
state_lock.acquire()
channel = state['channel']
amqp_exchange = session_profile['amqp_exchange'] amqp_exchange = session_profile['amqp_exchange']
# check first that sender didnt forget about reply to and corr id # check first that sender didnt forget about reply to and corr id
...@@ -155,6 +164,7 @@ def amqp_request(channel, request_message, component_id): ...@@ -155,6 +164,7 @@ def amqp_request(channel, request_message, component_id):
finally: finally:
# clean up # clean up
channel.queue_delete(reply_queue_name) channel.queue_delete(reply_queue_name)
state_lock.release()
return response return response
...@@ -168,6 +178,7 @@ def publish_message(message): ...@@ -168,6 +178,7 @@ def publish_message(message):
for i in range(1, 4): for i in range(1, 4):
try: try:
state_lock.acquire()
state['channel'].basic_publish( state['channel'].basic_publish(
exchange=session_profile['amqp_exchange'], exchange=session_profile['amqp_exchange'],
routing_key=message.routing_key, routing_key=message.routing_key,
...@@ -181,6 +192,9 @@ def publish_message(message): ...@@ -181,6 +192,9 @@ def publish_message(message):
_echo_error('Unexpected connection closed, retrying %s/%s' % (i, 4)) _echo_error('Unexpected connection closed, retrying %s/%s' % (i, 4))
_set_up_connection() _set_up_connection()
finally:
state_lock.acquire()
@click.group() @click.group()
def cli(): def cli():
...@@ -212,11 +226,15 @@ def repl(): ...@@ -212,11 +226,15 @@ def repl():
@cli.command() @cli.command()
def connect(): @click.option('-ll', '--lazy-listener',
is_flag=True,
default=False,
help="lazy-listener doest perform convertion to Messages class")
def connect(lazy_listener):
""" """
Connect to an AMQP session and start consuming messages Connect to an AMQP session and start consuming messages
""" """
_set_up_connection() _set_up_connection(lazy_listener=lazy_listener)
@cli.command() @cli.command()
...@@ -238,16 +256,11 @@ def download_network_traces(): ...@@ -238,16 +256,11 @@ def download_network_traces():
_handle_get_testcase_list() _handle_get_testcase_list()
ls = state['tc_list'].copy() ls = state['tc_list'].copy()
try:
channel = state['connection'].channel()
except Exception as e:
_echo_error(e)
for tc_item in ls: for tc_item in ls:
try: try:
tc_id = tc_item['testcase_id'] tc_id = tc_item['testcase_id']
msg = MsgSniffingGetCapture(capture_id=tc_id) msg = MsgSniffingGetCapture(capture_id=tc_id)
response = amqp_request(channel, msg, COMPONENT_ID) response = amqp_request(msg, COMPONENT_ID)
if response.ok: if response.ok:
save_pcap_from_base64(response.filename, response.value, TEMP_DIR) save_pcap_from_base64(response.filename, response.value, TEMP_DIR)
...@@ -298,11 +311,10 @@ def _handle_get_testcase_list(): ...@@ -298,11 +311,10 @@ def _handle_get_testcase_list():
# requires testing tool to implement GetTestCases feature, see MsgTestSuiteGetTestCases # requires testing tool to implement GetTestCases feature, see MsgTestSuiteGetTestCases
if _connection_ok(): if _connection_ok():
temp_channel = state['connection'].channel()
request_message = MsgTestSuiteGetTestCases() request_message = MsgTestSuiteGetTestCases()
try: try:
testcases_list_reponse = amqp_request(temp_channel, request_message, COMPONENT_ID) testcases_list_reponse = amqp_request(request_message, COMPONENT_ID)
except Exception as e: except Exception as e:
_echo_error('Is testing tool up?') _echo_error('Is testing tool up?')
_echo_error(e) _echo_error(e)
...@@ -494,6 +506,13 @@ def enter_debug_context(): ...@@ -494,6 +506,13 @@ def enter_debug_context():
""" """
_send_configuration_default_message_for_coap_testsuite() _send_configuration_default_message_for_coap_testsuite()
@cli.command()
def _get_session_configuration_from_ui():
"""
Get session config from UI
"""
_get_session_configuration()
@cli.command() @cli.command()
@click.argument('testcase_id') @click.argument('testcase_id')
def _testcase_skip(testcase_id): def _testcase_skip(testcase_id):
...@@ -573,11 +592,10 @@ def get_session_status(): ...@@ -573,11 +592,10 @@ def get_session_status():
# requires testing tool to implement GetStatus feature, see MsgTestSuiteGetStatus # requires testing tool to implement GetStatus feature, see MsgTestSuiteGetStatus
if _connection_ok(): if _connection_ok():
temp_channel = state['connection'].channel()
request_message = MsgTestSuiteGetStatus() request_message = MsgTestSuiteGetStatus()
try: try:
status_resp = amqp_request(temp_channel, request_message, COMPONENT_ID) status_resp = amqp_request(request_message, COMPONENT_ID)
except Exception as e: except Exception as e:
_echo_error('Is testing tool up?') _echo_error('Is testing tool up?')
_echo_error(e) _echo_error(e)
...@@ -634,14 +652,14 @@ def _echo_context(): ...@@ -634,14 +652,14 @@ def _echo_context():
_echo_list_as_table(table) _echo_list_as_table(table)
def _set_up_connection(): def _set_up_connection(lazy_listener=False):
global state
# conn for repl publisher # conn for repl publisher
try: try:
retries_left = CONNECTION_SETUP_RETRIES retries_left = CONNECTION_SETUP_RETRIES
state_lock.acquire()
while retries_left > 0: while retries_left > 0:
try: try:
state['connection'] = pika.BlockingConnection(pika.URLParameters(session_profile['amqp_url'])) state['connection'] = pika.BlockingConnection(pika.URLParameters(session_profile['amqp_url']))
state['channel'] = state['connection'].channel() state['channel'] = state['connection'].channel()
break break
...@@ -656,6 +674,9 @@ def _set_up_connection(): ...@@ -656,6 +674,9 @@ def _set_up_connection():
state['channel'] = None state['channel'] = None
return return
finally:
state_lock.release()
# note we have a separate conn for amqp listener (each pika threads needs a different connection) # note we have a separate conn for amqp listener (each pika threads needs a different connection)
if 'amqp_listener_thread' in state and state['amqp_listener_thread'] is not None: if 'amqp_listener_thread' in state and state['amqp_listener_thread'] is not None:
_echo_log_message('stopping amqp listener thread') _echo_log_message('stopping amqp listener thread')
...@@ -666,12 +687,22 @@ def _set_up_connection(): ...@@ -666,12 +687,22 @@ def _set_up_connection():
_echo_log_message('amqp listener thread doesnt want to stop, lets terminate it..') _echo_log_message('amqp listener thread doesnt want to stop, lets terminate it..')
th.terminate() th.terminate()
amqp_listener_thread = AmqpListener( if lazy_listener:
amqp_url=session_profile['amqp_url'], amqp_listener_thread = AmqpListener(
amqp_exchange=session_profile['amqp_exchange'], amqp_url=session_profile['amqp_url'],
callback=_message_handler, amqp_exchange=session_profile['amqp_exchange'],
topics=DEFAULT_TOPIC_SUBSCRIPTIONS, callback=None,
) topics=DEFAULT_TOPIC_SUBSCRIPTIONS,
use_message_typing=False,
)
else:
amqp_listener_thread = AmqpListener(
amqp_url=session_profile['amqp_url'],
amqp_exchange=session_profile['amqp_exchange'],
callback=_message_handler,
topics=DEFAULT_TOPIC_SUBSCRIPTIONS,
use_message_typing=True,
)