MQTT refactoring

Author: litnimax

.env.example

@@ -4,4 +4,4 @@ ODOO_IP=10.18.0.2
 DB_IP=10.18.0.3
 NGINX_IP=10.18.0.4
 BROKER_IP=10.18.0.5
-MARABUNTA_FORCE_VERSION=1.0.0
+MMARABUNTA_FORCE_VERSION=1.0.0

.gitignore

@@ -7,3 +7,6 @@ local_conf.py
 pymakr.conf
 docker-compose.override.yml
 deploy/ansible-deployment
+odoo/src/odoo
+# Ignore .conf files from local bound conainer folder
+asterisk/etc/asterisk/

asterisk/Dockerfile

@@ -9,7 +9,8 @@ RUN apk update && \
 #Install optional tools
 RUN apk add tcpdump ethtool vlan iftop ngrep bash vim screen tmux mosquitto-clients
 
+RUN mkdir /var/log/supervisor
+
 COPY ./ /
 
-RUN pip install -r /etc/agent/requirements.txt && \
-    pip install -r /etc/asterisk/services/requirements.txt
+RUN pip install -r /services/requirements.txt

asterisk/agent/conf.py

@@ -11,24 +11,30 @@ supervisor.rpcinterface_factory = supervisor.rpcinterface:make_main_rpcinterface
 logfile=/var/log/supervisord.log
 logfile_maxbytes=50MB
 logfile_backups=10
-loglevel=info
+loglevel=debug
 nodaemon=true
+childlogdir=/var/log/supervisor
 
 [program:asterisk]
 command=asterisk -fpq
+redirect_stderr=true
 
 [program:mqtt_client]
-directory=/etc/agent
+directory=/services
 command=python mqtt_client.py
+redirect_stderr=true
 
 [program:agent]
-directory=/etc/agent
+directory=/services
 command=python asterisk_helper.py
+redirect_stderr=true
 
 [program:ami_broker]
-directory=/etc/asterisk/services
+directory=/services
 command=python ami_broker.py
+redirect_stderr=true
 
 [program:stasis_apps]
-directory=/etc/asterisk/services
+directory=/services
 command=python stasis_apps.py
+redirect_stderr=true

asterisk/agent/requirements.txt

@@ -9,6 +9,8 @@
 logging.basicConfig(level=logging.DEBUG)
 _logger = logging.getLogger(__name__)
 
+ASTERISK = '/usr/bin/asterisk'
+
 from conf import *
 
 class MyTermSocket(TermSocket):

asterisk/etc/agent/asterisk_helper.py

@@ -16,14 +16,24 @@
 POLL_RECONNECT_TIMEOUT = 5
 AMI_RELOAD_PAUSE = 2 # 2 seconds between AMI connection reload
 
+# Agent
+PORT = 8010
+ASTERISK = '/usr/sbin/asterisk'
+ASTERISK_ARGS = '-cr'
+ASTERISK_RECORDING_FOLDER = '/var/spool/asterisk/monitor'
+SSL_ENABLED = False
+SSL_CERT = './cert.pem'
+SSL_KEY = './privkey.pem'
+
+
+
 ASTERISK_HELPER_URL = 'http://localhost:8010'
 
 UPDATE_CDR_DELAY = 5 # Delay X sec before updating cdr to make sure it's in DB.
 UPDATE_CHANNEL_DELAY = 1 # Delay X sec before updating cdr to make sure it's in DB.
 RECORDING_DOWNLOAD_DELAY = 1 # Delay to let Asterisk close recorded file.
 
 LOG_CONSOLE = True
-LOG_FILE = os.path.join(os.path.dirname(__file__), 'services.log')
 LOG_LEVEL = 'DEBUG'
 
 try:
@@ -36,8 +46,6 @@
 HANDLRES = []
 if LOG_CONSOLE:
     HANDLRES.append('console')
-if LOG_FILE:
-    HANDLRES.append('file')
 
 LOGGING = {
     'version': 1,
@@ -57,18 +65,6 @@
 }
 
 
-# Log to file if log file is configured
-if LOG_FILE:
-    LOGGING['handlers']['file'] = {
-        'level': LOG_LEVEL,
-        'class': 'logging.handlers.RotatingFileHandler',
-        'formatter': 'verbose',
-        'filename': LOG_FILE,
-        'maxBytes': 10485760,
-        'backupCount': 5
-    }
-
-
 LOGGING['loggers'] = {
     # Disable requests library spam
     'requests': {

asterisk/etc/agent/conf.py

@@ -0,0 +1,83 @@
+import json
+import paho.mqtt.client as mqtt
+import re
+import socket
+import os
+import uuid
+
+# Define Variables
+AST_ETC_DIR= "/etc/asterisk"
+MQTT_HOST = "broker"
+MQTT_PORT = 1883
+MQTT_KEEPALIVE_INTERVAL = 45
+HOSTNAME = socket.gethostname()
+
+
+
+class Client():
+
+    def __init__(self):
+        self.uid = os.environ.get('UID') or '{}'.format(uuid.getnode())
+        print 'UID: ' + self.uid
+        self.mqtt_client = mqtt.Client(client_id=self.uid)
+        self.mqtt_client.on_connect = self.on_connect
+        self.mqtt_client.on_message = self.on_message
+        self.mqtt_client.on_publish = self.on_publish
+        self.mqtt_client.on_subscribe = self.on_subscribe
+        self.mqtt_client.on_log = self.on_log
+        # Subscribe to myself
+        self.mqtt_client.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEPALIVE_INTERVAL)
+        self.mqtt_client.subscribe('/asterisk/' + self.uid + '/#', 0)
+        self.mqtt_client.loop_forever()
+
+
+
+    def on_connect(self, client, userdata, flags, rc):
+        print("rc: " + str(rc))
+
+
+    def on_message(self, client, userdata, msg):
+        print(msg.info, msg.dup, msg.mid, msg.qos, msg.retain, msg.state, msg.timestamp, msg.topic)
+        found = re.search('^/asterisk/{}/(.+)$'.format(self.uid), msg.topic)
+        if not found:
+            print('Error: topic {} not understood.'.format(msg.topic))
+        event_handler = getattr(self, 'on_' + found.group(1), self.handler_not_found)
+        event_handler(client, userdata, msg)
+
+    def on_publish(self, client, obj, mid):
+        print("mid: " + str(mid))
+
+
+    def on_subscribe(self, client, obj, mid, granted_qos):
+        print("Subscribed: " + str(mid) + " " + str(granted_qos))
+
+
+    def on_log(self, client, obj, level, string):
+        print(string)
+
+
+    def handler_not_found(self, client, userdata, msg):
+        print 'Topic {} handler not found.'.format(msg.topic)
+
+
+    def _extract_message(self, payload):
+        try:
+            msg = json.loads(payload)
+            return msg
+        except ValueError as e:
+            print(e.message, ': ', payload)
+
+
+    def on_file(self, client, userdata, msg):
+        data = self._extract_message(msg.payload)
+        file_name = data.get('FileName')
+        dest_folder = data.get('DestinationFolder')
+        filename = os.path.join(dest_folder, file_name)
+        print 'Updating file %s' % filename
+        with open(filename, 'w') as f:
+            f.write(data.get('Content'))
+
+
+
+# Initiate MQTT Client
+mqttc = Client()

asterisk/etc/agent/mqtt_client.py

@@ -2,8 +2,9 @@ git+https://github.com/digium/swagger-py.git
 ari
 odoorpc
 py-Asterisk
+terminado
 
 #next dependencies are installed with aline's apk but left here for reference
-gevent
-requests
-setproctitle
+#gevent
+#requests
+#setproctitle

asterisk/etc/agent/requirements.txt

@@ -14,4 +14,5 @@ services:
 
   asterisk:
     volumes:
-      - "./asterisk/etc/asterisk/services:/etc/asterisk/services"
+      - "./asterisk/etc/:/etc/"
+      - "./asterisk/services/:/services/"