lxbfYeaa/BookConverter
Archived
Template
1
0
Fork 0
forked from LiveCarta/BookConverter
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix consumer for run with queue

Browse Source
This commit is contained in:
Jeniamakarchik
2020-04-10 17:02:39 +03:00
parent 86a2e0876d
commit 66fb1cb5d6
1 changed files with 32 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
src/consumer.py
View File

@@ -92,36 +92,36 @@ def local_run(filename):
if __name__ == '__main__': if __name__ == '__main__':
# folder_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) folder_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# # config_path = Path(os.path.join(folder_path, "config/config.json")) # config_path = Path(os.path.join(folder_path, "config/config.json"))
# config_path = Path(os.path.join(folder_path, "config/queue_config.json")) config_path = Path(os.path.join(folder_path, "config/queue_config.json"))
# with open(config_path, "r") as f: with open(config_path, "r") as f:
# conf_param = json.load(f) conf_param = json.load(f)
#
# host = conf_param.get('host') or pika.ConnectionParameters().DEFAULT_HOST
# port = conf_param.get('port') or pika.ConnectionParameters().DEFAULT_PORT
#
# credentials = pika.PlainCredentials(username=conf_param['username'], password=conf_param['password'])
# parameters = pika.ConnectionParameters(host=host, port=port, credentials=credentials)
# connection = pika.BlockingConnection(parameters)
# channel = connection.channel()
#
# logger = configure_file_logger('consumer', logging_format='%(asctime)s - %(levelname)s - %(message)s')
#
# try:
# channel.queue_declare(queue=conf_param['queue'], durable=True, arguments={'x-max-priority': 10})
# except ValueError as exc:
# logger.log(logging.ERROR, f'Queue {conf_param["queue"]} is not declared.')
# raise exc
#
# acs = Access()
# channel.basic_consume(queue=conf_param['queue'],
# auto_ack=True,
# on_message_callback=partial(callback, access=acs, logger=logger))
# logger.info('Connection has been established.')
# print('Waiting for messages...')
# logger.info('Waiting for messages...')
#
# channel.start_consuming()
local_run('music') host = conf_param.get('host') or pika.ConnectionParameters().DEFAULT_HOST
port = conf_param.get('port') or pika.ConnectionParameters().DEFAULT_PORT
credentials = pika.PlainCredentials(username=conf_param['username'], password=conf_param['password'])
parameters = pika.ConnectionParameters(host=host, port=port, credentials=credentials)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
logger = configure_file_logger('consumer', logging_format='%(asctime)s - %(levelname)s - %(message)s')
try:
channel.queue_declare(queue=conf_param['queue'], durable=True, arguments={'x-max-priority': 10})
except ValueError as exc:
logger.log(logging.ERROR, f'Queue {conf_param["queue"]} is not declared.')
raise exc
acs = Access()
channel.basic_consume(queue=conf_param['queue'],
auto_ack=True,
on_message_callback=partial(callback, access=acs, logger=logger))
logger.info('Connection has been established.')
print('Waiting for messages...')
logger.info('Waiting for messages...')
channel.start_consuming()
# local_run('quote')