Developing Azira: From Concept to Diverse Applications

The Genesis of Azira:

Over the weekend, I built Azira an application designed to enable users to subscribe and receive crypto tokens almost instantaneously. Alongside this, I developed a Python library AziraClient to allow a seamless interaction with it. Here, I delve into the initial stages of Azira’s development. Currently, Azira operates on a set of dummy data, which was crucial for testing its functionality. This phase was not about a full-scale launch but rather a proof of concept to validate the core idea.

This article talks about my implementation process of Azira and the two possible directions the project would morph into.

The entire code implementation of this project and library are available:

• Application: https://github.com/manny-uncharted/Azira
• Library: https://pypi.org/project/aziraclient/

Proof of Concept — A Success Story:

I discuss how the concept of Azira translated into a tangible and successful implementation. The results were promising, indicating the application’s potential effectiveness in the real world. The project was built with Python, ZeroMQ, FastAPI, WebSockets, and SQLAlchemy. I’ll start with an overview of the technologies, the motivation for the project, the challenges, and possible improvements.

The guide would require some experience with the following technologies.

• Python: The programming language that came to mind in the spur of the moment 😁. You’re free to explore building something similar in a language you’re most comfortable with.
• FastAPI: A web framework for building web applications in Python.
• ZeroMQ: An open-source universal messaging library.
• Websockets: A bidirectional communication protocol that can send the data from the client to the server or from the server to the client by reusing the established connection channel.
• SQLAlchemy: An easy-to-use database ORM manager that makes it easy to work with different databases.

Initial Approach

The project started with the notion of it being a simple implementation without even the thought of creating the library. Like most developers with just a rough sketch of how things would pan out drawn out in my head and some rough understanding of zeroMQ, I embarked on writing my first line of code.

In the above sketch, the following were considered:

• ZeroMQ server: Acts as a message bus that continuously streams the full tokens available to users.
• Web application: The interface between the user and the rest of the application using WebSockets.
• Connection Manager: A custom module I wrote to handle interactions between the Web application and the ZeroMQ server.

I noticed some faults when I began testing the initial approach:

I was unable to receive any message from the ZeroMQ server, and I discovered there was a problem with the communication between my web server and ZeroMQ. ZeroMQ is a messaging library that allows you to connect using a couple of pair approaches, you can have the option of using two communication protocols. :

• req/rep: this stands for ‘request’ and ‘response’. The REQ/REP sockets are Synchronous sockets. This means that they can only talk to one peer at a time
• pub/sub: this stands for ‘publish’ and ‘subscribe’. The basic idea of a publish/subscribe model is that a PUB socket pushes messages out, and all the associated SUB sockets receive these messages. This communication is strictly one-way, there are no responses or acknowledgments sent by the SUB sockets.

I realized that only the publish connection was initialized and the subscribe connection was neglected. This prompted a redesign of the approach shown below:

Evolved Design

Evolved design

Implementation Insights

This section is a deep dive into my process of developing Azira. I share challenges, breakthroughs, and the technical intricacies involved in bringing this idea to life.

If you would like to implement this project on your own, here is a tutorial walkthrough of what I did.

Note: A lot of the implementations here might not be best practices, as it was a weekend build, which I would slowly ramp up.

The implementation is in two stages:

• Web application Implementation.
• Library Implementation

Web application Implementation

Following the evolved design we now have the following:

• ZeroMQ server: Acts as a message bus that continuously streams the full tokens available to users.
• Messaging Listener: Acts as an interface between the ZeroMQ server and the connection manager. It creates a connection point between the ZeroMQ server and sends the tokens being streamed to the connection manager, to be accessed by the FastAPI server.
• Authentication system: To handle user creation and login to the service.
• FastAPI server: The interface between the user and the rest of the application using WebSockets.
• Connection Manager: A custom module I wrote to handle interactions between the Web application and the ZeroMQ server.

Building the ZeroMQ server

Let’s start by creating our virtual environment to start coding:

mkdir azira
cd azira
mkvirtualenv azira

Moving on to installing our required libraries and saving it to a requirements file:

pip install fastapi websockets requests uvicorn zmq sqlalchemy celery alembic jwt
pip freeze > requirements.txt

The zeroMQ server was built using the pub/sub communication protocol, which means it can only send information and doesn’t allow bi-directional communication from the sub connections. This ensures that it can stream the tokens' data at high throughput without interruptions.

Let’s create our zeroMQ server file.

mkdir notifications
cd notifications
touch messaging_bq.py

import zmq 
import random 
import json 
import time
import notifier
import threading

class zmqConn():
    def __init__(self):
        self.pubsocket = self.__zmq_config(5556)
        self.functions = {
                            "addto_Feed" : self.addto_Feed,
                            "terminate" : self.terminate
                          }
        self.tokens = ["NSE:26009", "NSE:26000", "NSE:212", "NSE:230"]
        self.load = True
        self.runserver = True
        threading.Thread(target = self.load_data).start()
        self.server_setup("5558")
        
    def __zmq_config(self, port):
        context = zmq.Context()
        socket = context.socket(zmq.PUB)
        socket.bind("tcp://*:%s" % port)
        return socket
    
    def addto_Feed(self, values):
        payload = values.get('payload')
        if payload == None: 
            raise Exception("payload missing")
        else: 
            tokens = payload['tokens']
            for t in tokens: 
                if t not in self.tokens:
                    self.tokens.append(t)
                
    def load_data(self):
        while self.load:
            self.get_ltp()
            time.sleep(0.1)
            
    def get_ltp(self):
        for i in self.tokens:
            ltp = random.randint(500, 520)
            self.pubsocket.send_string("{topic} & {marketdata}".format(topic = i, marketdata = json.dumps({"token" : i, "ltp" : ltp, "bestbidprice" : ltp - 0.5,
                                                                                                           "bestaskprice" : ltp + 0.5})))
            time.sleep(0.2)
    
    def exit_app(self, timer = 2):
        time.sleep(timer)
        self.runserver = False
        self.load = False
        try: 
            self.context.destroy()
        except:pass

    def terminate(self, values):
        t1 = threading.Thread(target = self.exit_app).start()
        values.update({"exit" : "True"})
        return values
    
    def server_setup(self, serverport): 
        self.context = zmq.Context()
        socket = self.context.socket(zmq.REP)
        socket.bind("tcp://*:%s" % serverport)
        while self.runserver : 
            message = socket.recv().decode()
            try : 
                print(message)
                notifier.notification.delay(message)
                message = json.loads(message)
                message.update({"port" : serverport})
                try :  
                    resprecv = self.functions[message['function']](message)
                    resp = {"status" : True, "error" : False, "data" : [resprecv], "message" : "Data Received."}
                except Exception as e :
                    resp = {"status" : False, "error" : True,"message" : str(e)}
                socket.send(json.dumps(resp).encode())
            except Exception as e : 
                socket.send(json.dumps({"status" : False, "error" : True, "message" : str(e)}).encode())
            time.sleep(0.1)

zq = zmqConn()

To start the ZeroMQ server run:

python messaging_bq.py

Implementing the Messaging Listener

From above I discussed that the communication protocol used in my implementation was the “pub/sub”. Here the listener is made to subscribe to all tokens in the stream and hand them over to the Connection Manager. As this was the fix needed to complete the handshake missing from the previous approach.

Here we would be implementing the listener:

touch listener.py

import zmq
import asyncio
import logging
from notifications.websocket_manager import manager

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def zmq_listener():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect("tcp://localhost:5556")
    socket.setsockopt_string(zmq.SUBSCRIBE, '')  # Subscribe to all messages initially

    while True:
        try:
            message = await asyncio.to_thread(socket.recv_string)
            # Log received message
            # logger.info(f"Received message from ZeroMQ: {message}")
            token, data = message.split(' & ')
            manager.update_available_tokens(token)
            await manager.broadcast_to_subscribers(token, data)
        except zmq.Again:
            # No message was available, wait shortly and try again
            logger.info("No message was available, wait shortly and try again")
            await manager.send_error_message("No message was available, wait shortly and try again later")
            await asyncio.sleep(0.01)
        except Exception as e:
            logger.error(f"Error in zmq_listener: {e}")
            await asyncio.sleep(1)  # Prevent a tight loop on continuous error

Implementing the Connection Manager

The task of the connection manager is to efficiently handle requests between the web server and the listener. As it was set to terminate any client connection that exceeds a maximum of 3.

This was done to ensure that the service remains available to everyone. As the project morphs into the two directions being considered, this service would be revised to efficiently handle more than 3 connections from a user.

touch websocket_manager.py

from fastapi import WebSocket, HTTPException


MAX_CONNECTIONS_PER_ACCOUNT = 3

class ConnectionManager:

    def __init__(self):
        self.active_connections: dict = {}
        self.available_tokens: set = set() 

    async def connect(self, websocket: WebSocket, username: str):
        if self.active_connections.get(username, None) and len(self.active_connections[username]) >= MAX_CONNECTIONS_PER_ACCOUNT:
            await websocket.close(code=1008)  # Close connection with policy violation error
            return False
        await websocket.accept()
        if not self.active_connections.get(username):
            self.active_connections[username] = []
        self.active_connections[username].append({
            "websocket": websocket,
            "subscriptions": set()
        })
        return True
    def update_available_tokens(self, token: str):
        self.available_tokens.add(token)

    def is_token_available(self, token: str) -> bool:
        return token in self.available_tokens

    def disconnect(self, websocket: WebSocket, username: str):
        if self.active_connections.get(username):
            self.active_connections[username] = [
                connection for connection in self.active_connections[username]
                if connection["websocket"] != websocket
            ]
            if not self.active_connections[username]:
                del self.active_connections[username]

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def send_error_message(self, message:str):
        for username, connections in self.active_connections.items():
            for connection in connections:
                    await connection["websocket"].send_text(message)

    async def broadcast_to_subscribers(self, token: str, message: str):
        for username, connections in self.active_connections.items():
            for connection in connections:
                if token in connection["subscriptions"]:
                    await connection["websocket"].send_text(message)

manager = ConnectionManager()

The next step is to implement the Authentication system

Implementing the Authentication Service and Fastapi Server

This is split into two:

• Authentication Service (registration & login)
• WebSocket Endpoint

Let’s discuss the implementation of the registration service, at this step, we need to create our FastAPI application.

In the root folder ./azira, let’s create the following folders

mkdir app
mkdir crud
mkdir models
mkdir schemas
mkdir routes
mkdir middleware
mkdir db

I always like to structure my projects in order to ensure that they are readable and easy to understand, respecting the clean code approach.

• The app directory: contains the main file to run and start the FastAPI server and this is where I’d link all my routes, the web socket routes, and authentication routes.

cd app
touch main.py

from fastapi import FastAPI
import asyncio
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

from middleware.middleware import ProcessTimeMiddleware #, DBSessionMiddleware

from routes import users, trigger_router
from notifications.messaging_bq import zmqConn

origins = [
    "*"
]

app = FastAPI(debug=True)

app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
# app.add_middleware(DBSessionMiddleware)
app.add_middleware(ProcessTimeMiddleware)

# Authentication route
app.include_router(users.router, prefix="/api/v1/users")

# Websocket route
app.include_router(trigger_router.trigger)

• The crud directory: contains all the orm calls to my database following the CRUD pattern. Here I define CRUD functions for each route/endpoint that would be made using SQLAlchemy to the database.

cd crud
touch users.py

#! usr/bin/python3
# crud/users

from sqlalchemy.orm import Session
from models.users import User
from schemas.users import UserCreate
from utils.auth_jwt import create_hash_password, verify_password, create_access_token

def create_user(db: Session, user: UserCreate):
    hashed_password = create_hash_password(user.password)
    db_user = User(username=user.username, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(User).offset(skip).limit(limit).all()

def update_user(db: Session, user_id: int, user: UserCreate):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user:
        db_user.username = user.username
        db_user.hashed_password = create_hash_password(user.password)
        db.commit()
        db.refresh(db_user)
        return db_user

I know a lot of people might object to using ORMs as you might tag it as slow and a lazy man’s approach to development or talk about the amount of join queries that happen under the hood, yeah yeah, remember this was just a weekend project.

• The models directory: contains the model definitions, for the tables I need for this project. To keep things low, I only accounted for the authentication system.

cd models
touch users.py

#! usr/bin/python3
# models/users

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from db.db import Base


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    tokens = relationship("Token", back_populates="user", uselist=False)

• The schemas directory: contains the input schema definitions, here I made use of the Pydantic schema to ensure that my inputs have a type checking associated with them without having to worry about wrong inputs being sent through.

cd schemas
touch users.py

#!usr/bin/python3
# schemas/users
from pydantic import BaseModel
from typing import Optional

class UserBase(BaseModel):
    username: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: Optional[int] = None

    class Config:
        orm_mode = True

• The routes directory: contains the application routes with the authentication routes/endpoint separated from the WebSocket route.

cd routes
touch users.py
touch trigger_router.py

nano users.py

In the users.py routes file, we define the endpoints needed for registration and login.

#! usr/bin/python3
# routes/users.py

from fastapi import APIRouter, Depends, HTTPException, status, Cookie, Response, BackgroundTasks
from sqlalchemy.orm import Session
from datetime import datetime, timedelta


# local imports
from models.users import User
from schemas.users import UserCreate, User as UserSchema
from schemas.tokens import TokenCreate, Token as TokenSchema
from models.tokens import Token
from crud.users import create_user, get_user
from crud.tokens import create_user_token, get_token_by_user_id
from utils.auth_jwt import create_hash_password, create_access_token, verify_password, ACCESS_TOKEN_EXPIRE_MINUTES
from middleware.middleware import get_db
import db.db as DB
from utils.bg_tasks import update_user_token_every_30_mins

DB.Base.metadata.create_all(bind=DB.engine)

router = APIRouter()

@router.post('/register')
def register_user(user: UserCreate, response: Response, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.username == user.username).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Username already registered")

    # create jwt token
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=access_token_expires
    )

    # Set cookies with expiry
    response.set_cookie(key="access_token", value=f"Bearer {access_token}", expires=ACCESS_TOKEN_EXPIRE_MINUTES*60)
    response.set_cookie(key="client_id", value=user.username, expires=ACCESS_TOKEN_EXPIRE_MINUTES*60)
    
    try:
        created_user = create_user(db=db, user=user)
        create_user_token(db=db, token=access_token, user_id=created_user.id)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Error creating user: {str(e)}")

    return {"message": "User registered successfully"}


@router.post('/login')
def get_user_and_login(user: UserCreate, response: Response, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    # Check if the user exists in the database
    db_user = db.query(User).filter(User.username == user.username).first()
    if not db_user:
        raise HTTPException(status_code=400, detail="Invalid username or password")

    # Verify the password
    pass_verify = verify_password(user.password, db_user.hashed_password)
    if not pass_verify:
        raise HTTPException(status_code=400, detail="Invalid username or password")

    # If both username and password are correct, proceed with login
    try:
        user_access_token = get_token_by_user_id(db=db, user_id=db_user.id)
        # Start the background task
        # background_tasks.add_task(update_user_token_every_30_mins, db, db_user.id)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Error during login: {str(e)}")

    return {
        "message": f"User {db_user.username} logged in successfully",
        "access_token": user_access_token,
        "token_type": "bearer"
    }

In the trigger_router.py routes file, we define the endpoints needed for registration and login.

nano trigger_router.py

from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends, HTTPException
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
from sqlalchemy.orm import Session
import zmq
import jwt
import time
import asyncio

#Local imports
from notifications.websocket_manager import manager
from notifications.listener import zmq_listener
from utils.auth_jwt import SECRET_KEY, ALGORITHM
from crud.tokens import get_token_by_user_id
from crud.users import get_user
from middleware.middleware import get_db
from models.users import User
from models.tokens import Token


trigger = APIRouter(prefix="/api/trigger", tags=["trigger_in"])

@trigger.websocket("/ws/{username}")
async def websocket_endpoint(websocket: WebSocket, username: str, db: Session = Depends(get_db)):
    # Extract the JWT token from the headers
    token = None
    for key, value in websocket.headers.items():
        if key.lower() == "authorization":
            token = value.split(" ")[1]  # Assumes the header is in the format "Bearer <token>"
            break

    if not token:
        await websocket.close(code=1008)
        return
    print(f"Token: {token}")
    db_user = db.query(User).filter(User.username == username).first()
    if not db_user:
        raise HTTPException(status_code=401, detail="Invalid username")

    token_user = db.query(Token).filter(Token.access_token == token).first()
    print(f"User token from db: {token_user}")
    if not token_user:
        raise HTTPException(status_code=401, detail="Invalid token")


    # Try to connect and enforce connection limits
    if not await manager.connect(websocket, username):
        return

    try:
        while True:
            data = await websocket.receive_text()
            action, tokens = data.split('|')
            tokens = tokens.split(',')
            
            if action == "subscribe":
                missing_tokens = []
                for token in tokens:
                    # Check if the token is available in the stream (implement this check as needed)
                    if not manager.is_token_available(token):  # You need to implement this function
                        missing_tokens.append(token)
                    else:
                        connection = next((c for c in manager.active_connections[username] if c["websocket"] == websocket), None)
                        if connection:
                            connection["subscriptions"].add(token)

                if missing_tokens:
                    await manager.send_personal_message(f"Tokens not found: {', '.join(missing_tokens)}", websocket)
                    await websocket.close(code=1008)  # Close the connection
                    return
                else:
                    await manager.send_personal_message(f"Subscribed to tokens: {', '.join(tokens)}", websocket)
            
            elif action == "unsubscribe":
                for token in tokens:
                    connection = next((c for c in manager.active_connections[username] if c["websocket"] == websocket), None)
                    if connection:
                        connection["subscriptions"].discard(token)
                await manager.send_personal_message(f"Unsubscribed from tokens: {', '.join(tokens)}", websocket)
            
            else:
                await manager.send_personal_message(f"Unknown action: {action}", websocket)

    except WebSocketDisconnect:
        manager.disconnect(websocket, username)
    except Exception as e:
        await manager.send_personal_message(f"Error: {str(e)}", websocket)
        manager.disconnect(websocket, username)

@trigger.on_event("startup")
async def startup_event():
    asyncio.create_task(zmq_listener())

• The middleware directory: contains helper functions that I need to run in the background, like a function that calculates the server request-response time, this is a useful metric that comes in handy when you notice the application starts slowing down, it can help identify which of the endpoints is causing this issue.

cd middleware
touch middleware.py

nano middleware.py

import time
from fastapi import Request, HTTPException, Depends
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from db.db import SessionLocal

# Dependency to get DB session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class ProcessTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

• The db directory: contains my database configuration file.

cd db
touch db.py

nano db.py

import time
from fastapi import Request, HTTPException, Depends
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from db.db import SessionLocal

# Dependency to get DB session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class ProcessTimeMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

• The utils directory: contains utility functions needed for the application to function well.

cd utils.py
touch auth_jwt.py

nano auth_jwt.py

#! usr/bin/python3
# utils/auth_jwt
## Jwt implementation

import os
import bcrypt
import jwt
import binascii
from datetime import datetime, timedelta
from dotenv import load_dotenv
load_dotenv()

SECRET_KEY = os.environ.get("SECRET")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Utility function to hash passwords
def create_hash_password(password: str):
    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())

# Utility function to verify passwords
def verify_password(plain_password, hashed_password):
    if not isinstance(plain_password, bytes):
        plain_password = plain_password.encode('utf-8')

    # Convert the hexadecimal string to bytes if necessary
    if isinstance(hashed_password, str) and hashed_password.startswith("\\x"):
        hashed_password = binascii.unhexlify(hashed_password[2:])

    validator = bcrypt.checkpw(plain_password, hashed_password)
    print(f"Validator check: {validator}")
    return validator


# Utility function to generate JWT tokens
def create_access_token(*, data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

Testing the application

• In your main directory start the web application using:

cd azira
uvicorn app.main:app

To start the ZeroMQ server:

cd notifications
python messaging_bq.py

• You need to install the AziraClient in order to test-run the application.

pip install aziraclient

• Let’s create our test folder

mkdir tests
cd tests
touch test.py
nano test.py

import asyncio

# Importing the necessary modules from the library
from aziraclient.message_bus.message_bus_tester import ZMQTestSubscriber
from aziraclient.subscription.subscription import SubscribeToToken
from aziraclient.auth.auth_client import AuthClient

def main():
    """
    Main function to demonstrate the usage of the library modules.
    """
    while True:
        print("\nSelect an option to test:")
        print("1. Register User")
        print("2. Login User")
        print("3. Test WebSocket Subscription")
        print("4. Test ZMQ Subscriber")
        print("5. Exit")
        choice = input("Enter your choice: ")

        if choice == '1':
            # User registration
            username = input("Enter username for registration: ")
            password = input("Enter password for registration: ")
            #If you're running the application locally you can use this to define your app url
            auth = AuthClient(url="http://localhost:8000")
            print("Registering user...")
            register = auth.register_user(username, password=password)
            print("User registration response:", register)

        elif choice == '2':
            # User login
            username = input("Enter username for login: ")
            password = input("Enter password for login: ")
            auth = AuthClient()
            print("Logging in user...")
            login = auth.login_user(username, password)
            print("User login response:", login)

        elif choice == '3':
            # WebSocket subscription testing
            username = input("Enter your username: ")
            jwt_token = input("Enter your JWT token: ")
            action = input("Enter an action (subscribe or unsubscribe): ")
            token = input("Enter the name of the token you would like to subscribe to or unsubscribe from: ")
            tester = SubscribeToToken(username, jwt_token, action, token)
            asyncio.get_event_loop().run_until_complete(tester.test_connection())

        elif choice == '4':
            # ZMQTestSubscriber testing
            print("Starting ZMQTestSubscriber...")
            subscriber = ZMQTestSubscriber(tokens=["NSE:26009", "NSE:26000"])
            subscriber.subscribe()

        elif choice == '5':
            print("Exiting...")
            break

        else:
            print("Invalid choice. Please select a valid option.")

if __name__ == '__main__':
    main()

Run the above file using:

Looking Ahead — The Evolution of Azira:

Building on the success of the initial proof of concept, I outline my vision for transforming Azira. I propose two distinct directions for its evolution, each branching out into separate projects with unique applications and goals.

• First Direction: An application that grants users the ability to subscribe and receive crypto tokens data in near real-time. Its use case cuts across those who are in the crypto trading space to finance, as I would most likely extend it to work for assets trading.
• Second Direction: One that allows people to build distributed communications systems from simple messaging systems with little to no knowledge about them

As of the time of writing this blog post, the project would have begun its morphological process in the first direction, it’s approach is still very much similar to what was described here. But with more stable fixes and updates.

Conclusion

Reflecting on the journey of Azira, from a mere concept to a fully functional prototype. This wasn’t meant to be a tutorial, just a blog about a weekend project and a couple of ideas on how I intend to build it further. I know I didn’t talk about how I built the library, AziraClient, that would come in another post. I learned a lot about messaging systems here, and hope to work more with them in the future. I think I should try out more weekend builds 😁.