IraVoice Gateway Status API
This Python script runs a background service that connects to the IraCallScheduler NATS messaging subject and maintains the status of gateways in memory. It listens for gateway updates and allows an API to query the current status of a gateway.
Overview
The service performs three main tasks:
Connects securely to a NATS server using an NKey seed.
Listens for gateway status update events.
Responds to gateway status requests.
The status of each gateway is stored in memory while the service is running.
IraVoice emits a request_update_gateway_mapping message to the IraCallScheduler on iracallscheduler.request subject which contains the gateway name and also the status of it.
Example-
{"event_data":{"dialer_ip_address":"192.168.11.13","dialer_pod_id":"iravoice-0711662b-4e7a-46c7-af0c-bbcd96319e56","gateway_name":"airtel","gateway_status":"UP","usage":0},"event_name":"request_update_gateway_mapping"} |
The Python script subscribes to the iracallscheduler request subject and then waits for the request_update_gateway_mapping message and holds the gateway name and the status of it in memory.
Script attached below.
import asyncio
import json
import nats
async def main():
key_file_path = "/usr/local/epi/scripts/py/mysign.nk"
config_path = "/usr/local/epi/conf/common_config.json"
nats_url = "nats://127.0.0.1:4222"
try:
with open(config_path, "r") as f:
config_data = json.load(f)
if "nats_conf" in config_data and "nats_url" in config_data["nats_conf"]:
nats_url = config_data["nats_conf"]["nats_url"]
print(f"Loaded NATS URL from config: {nats_url}")
else:
print(f"Warning: 'nats_conf.nats_url' not found in {config_path}. Using default.")
except FileNotFoundError:
print(f"Error: Config file not found at {config_path}. Using default URL.")
except json.JSONDecodeError:
print(f"Error: Failed to parse JSON at {config_path}. Using default URL.")
try:
with open(key_file_path, "r") as f:
seed_key = f.read().strip()
except FileNotFoundError:
print(f"Error: Could not find the seed key file at {key_file_path}")
return
except PermissionError:
print(f"Error: Permission denied when trying to read {key_file_path}")
return
if not seed_key:
print("Error: The seed key file is empty.")
return
print("Connecting to NATS...")
try:
nc = await nats.connect(
nats_url,
nkeys_seed_str=seed_key
)
print("Authenticated and connected successfully.")
except Exception as e:
print(f"Failed to connect to NATS: {e}")
return
gateway_status_store = {}
async def update_handler(msg):
try:
payload = json.loads(msg.data.decode())
if payload.get("event_name") == "request_update_gateway_mapping":
name = payload.get("event_data", {}).get("gateway_name")
status = payload.get("event_data", {}).get("gateway_status")
if name and status:
gateway_status_store[name] = status
print(f"Recorded update: {name} is {status}")
except Exception as e:
print(f"Error parsing update: {e}")
await nc.subscribe("*.iracallscheduler.request", cb=update_handler)
async def request_handler(msg):
try:
payload = json.loads(msg.data.decode())
if payload.get("event_name") != "request_gateway_status":
await msg.respond(json.dumps({
"error": "Invalid event_name"
}).encode())
return
gateway_name = payload.get("event_data", {}).get("gateway_name")
if not gateway_name:
await msg.respond(json.dumps({
"error": "gateway_name is required"
}).encode())
return
status = gateway_status_store.get(gateway_name, "UNKNOWN")
response = json.dumps({
"gateway_name": gateway_name,
"status": status
})
await msg.respond(response.encode())
except Exception as e:
await msg.respond(json.dumps({
"error": f"Invalid JSON: {str(e)}"
}).encode())
await nc.subscribe("gateway.status.request", cb=request_handler)
print("Service running: Listening for requests on 'gateway.status.request'...")
while True:
await asyncio.sleep(1)
if __name__ == '__main__':
asyncio.run(main()) |
The above script also creates a custom NATS subject gateway.status.request which expects a request to be sent with the following format to the custom NATS subject.
Example-
{
"event_name": "request_gateway_status",
"event_data": {
"gateway_name":"airtel"
}
} |
Response to the API will be as follows.
{
"gateway_name": "airtel",
"status": "UP"
} |
You can use HTTP nats to take the http request and then send it to the custom NATS subject.
Example-
You can create a service to run the python script in the background. Attached below is the service file.
[Unit]
Description=Iravoice Gateway API script
After=syslog.target network.target network-online.target local-fs.target
[Service]
User=root
Group=root
WorkingDirectory=/usr/local/epi/scripts
ExecStart=/usr/local/epi/scripts/py/venv/bin/python3 /usr/local/epi/scripts/gateway_status_prog.py
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target |
