rabbitmq_ari_proxy.py

#!/usr/bin/python
import websocket
import time
import pika
import sys
import json
import requests
import os
import argparse

if (sys.version_info < (3, 0)):
    import thread as _thread

channel = None
credentials = None
args = None

def on_message(ws, message):
    channel.basic_publish(exchange=args.rabbitmq_outbound_exchange, routing_key=args.server_id, body=message)

def on_error(ws, error):
    print(error)

def on_close(ws):
    print("### closed ###")

def on_open(ws):
    print("Connected")

def consume_callback(ch, method, properties, body):
    request = json.loads(str(body))

    result_url = "http://" + args.base_url + "/ari" + request['resource'] + credentials
    if request['options'] != "":
        result_url = result_url + "&" + request['options']

    response = None

    if request['method'] == 'get':
        response = requests.get(result_url)
    elif request['method'] == 'post':
        response = requests.post(result_url)
    elif request['method'] == 'delete':
        response = requests.delete(result_url)
    elif request['method'] == 'put':
        response = requests.put(result_url)

    response_to_sender = json.dumps({"code": response.status_code, "response": response.text})

    ch.basic_publish(exchange='', routing_key=properties.reply_to, properties=pika.BasicProperties(correlation_id = properties.correlation_id),
        body=str(response_to_sender))

    ch.basic_ack(delivery_tag = method.delivery_tag)


def start_consuming():
    channel.start_consuming()

if __name__ == "__main__":

    parser = argparse.ArgumentParser(description='ARI proxy to rabbitmq')
    parser.add_argument('--base-url', type=str,
                    help='Asterisk base url', default="127.0.0.1:8088")
    parser.add_argument('--api-key', type=str,
                    help='ARI API key', default="root")
    parser.add_argument('--api-pass', type=str,
                    help='ARI API password', default="")
    parser.add_argument('--api-app', type=str,
                    help='ARI API application', default="ari")
    parser.add_argument('--rabbitmq-url', type=str,
                    help='Rabbitmq connection url', default="amqp://localhost")
    parser.add_argument('--rabbitmq-outbound-exchange', type=str,
                    help='Rabbitmq exchange for outbound messasages', default="ari_exchange_out")
    parser.add_argument('--rabbitmq-inbound-exchange', type=str,
                    help='Rabbitmq exchange for incoming messasages', default="ari_exchange_in")
    parser.add_argument('--server-id', type=str,
                    help='Routing key for rabbitmq', default=os.uname()[1])

    args = parser.parse_args()

    credentials = "?api_key=%s%s%s&app=%s" % (args.api_key, "%3A", args.api_pass, args.api_app)

    connection = pika.BlockingConnection(pika.URLParameters(args.rabbitmq_url))
    channel = connection.channel()
    channel.exchange_declare(exchange=args.rabbitmq_outbound_exchange, type='fanout', durable=True)
    channel.exchange_declare(exchange=args.rabbitmq_inbound_exchange, type='direct')

    result = channel.queue_declare(exclusive=True)
    direct_queue = result.method.queue

    channel.queue_bind(exchange=args.rabbitmq_inbound_exchange, queue=direct_queue, routing_key=args.server_id)
    channel.basic_consume(consume_callback, queue=direct_queue)


    websocket.enableTrace(False)
    ws = websocket.WebSocketApp("ws://" + args.base_url + "/ari/events" + credentials,
                                on_message = on_message,
                                on_error = on_error,
                                on_close = on_close,
                                on_open = on_open)
    _thread.start_new_thread(start_consuming, ())
    ws.run_forever()