diff --git a/ohpc_account_create.py b/ohpc_account_create.py index ee69364cfc859143ca6a560ba43469851b1f8d9c..80e80db97d854ebc3bf7c56ba7df3259586bf7b9 100644 --- a/ohpc_account_create.py +++ b/ohpc_account_create.py @@ -7,57 +7,48 @@ import subprocess import time import json from pwd import getpwnam +from rc_rmq import RCRMQ -hostname = socket.gethostname().split(".", 1)[0] -connect_host = rcfg.Server if hostname != rcfg.Server else "localhost" -queue_name = "ohpc_account_create" -duration = 2 +task = "ohpc_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 ohpc_account_create(ch, method, properties, body): msg = json.loads(body) print("Message received {}".format(msg)) username = msg['username'] + success = False try: subprocess.call(["sudo", "useradd", 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) + ch.basic_ack(delivery_tag=method.delivery_tag) msg['uid'] = getpwnam(username).pw_uid msg['gid'] = getpwnam(username).pw_gid - channel.basic_publish(exchange=rcfg.Exchange, routing_key='ood_account_create', body=json.dumps(msg)) - - -# ingest messages -channel.basic_consume(queue=queue_name, on_message_callback=ohpc_account_create) - -# initiate message ingestion -try: - channel.start_consuming() -except KeyboardInterrupt: - print("Disconnecting from broker.") - channel.stop_consuming() -connection.close() + # send confirm message + rc_rmq.publish_msg({ + 'routing_key': 'confirm.' + username, + 'msg': { + 'task': task, + 'success': success + } + }) + + if success: + # send create message to other agent + rc_rmq.publish_msg({ + 'routing_key': 'create.' + username, + 'msg': msg + }) + +print("Start Listening to queue: {}".format(task)) +rc_rmq.start_consume({ + 'queue': task, + 'routing_key': 'request.*', + 'cb': ohpc_account_create +})