diff --git a/ood_account_create.py b/ood_account_create.py index 2abaa23429cd41e4565a25e2055ec18920713328..332238620bd5f97d86b6fc54d1aefa728c1a26a1 100644 --- a/ood_account_create.py +++ b/ood_account_create.py @@ -6,32 +6,12 @@ import socket import subprocess import time import json +from rc_rmq import RCRMQ -hostname = socket.gethostname().split(".", 1)[0] -connect_host = rcfg.Server if hostname != rcfg.Server else "localhost" -queue_name = "ood_account_create" -duration = 2 +task = 'ood_account' -# Set up credentials to connect to RabbitMQ server -credentials = pika.PlainCredentials(rcfg.User, rcfg.Password) -parameters = pika.ConnectionParameters(connect_host, - rcfg.Port, - rcfg.VHost, - credentials) - -# Establish connection to RabbitMQ server -connection = pika.BlockingConnection(parameters) -channel = connection.channel() - -print("connection established. Listening for messages:") - -# create exchange to pass messages -channel.exchange_declare(exchange=rcfg.Exchange, exchange_type='direct') - -# creates a random name for the newly generated queue -result = channel.queue_declare(queue=queue_name, exclusive=False) - -channel.queue_bind(exchange=rcfg.Exchange, queue=queue_name, routing_key=queue_name) +# Instantiate rabbitmq object +rc_rmq = RCRMQ({'exchange': 'RegUsr', 'exchange_type': 'topic'}) def ood_account_create(ch, method, properties, body): msg = json.loads(body) @@ -39,26 +19,29 @@ def ood_account_create(ch, method, properties, body): username = msg['username'] user_uid = str(msg['uid']) user_gid = str(msg['gid']) + success = False try: subprocess.call(["sudo", "groupadd", "-r", "-g", user_gid, username]) subprocess.call(["sudo", "useradd", "-u", user_uid, "-g", user_gid, username]) - print("User {} has been added to {}".format(username, hostname)) + print("[{}]: User {} has been added".format(task, username)) + success = True except: print("Failed to create user") - channel.basic_ack(delivery_tag=method.delivery_tag) - - channel.basic_publish(exchange=rcfg.Exchange, routing_key='slurm_add_account', body=json.dumps(msg)) - - -# ingest messages -channel.basic_consume(queue=queue_name, on_message_callback=ood_account_create) - -# initiate message ingestion -try: - channel.start_consuming() -except KeyboardInterrupt: - print("Disconnecting from broker.") - channel.stop_consuming() - -connection.close() + ch.basic_ack(delivery_tag=method.delivery_tag) + + # send confirm message + rc_rmq.publish_msg({ + 'routing_key': 'confirm.' + username, + 'msg': { + 'task': task, + 'success': success + } + }) + +print("Start listening to queue: {}".format(task)) +rc_rmq.start_consume({ + 'queue': task, + 'routing_key': "create.*", + 'cb': ood_account_create +})