DEV Community: Vinícyus Brasil

Caching API responses with Redis in Zig

Vinícyus Brasil — Wed, 12 Mar 2025 02:10:16 +0000

Introduction

When fetching a database, our server must perform a READ operation. Every transation has a cost of time and money (computational cost in the cloud). One way of reducing this cost it's to cache the last record fetched so we don't have to perform another READ operation. This can help when the same record is requested multiple times.

To do this caching, we can use Redis, a advanced key-value store that works as a in memory database. It is really fast and combines with our choice for the backend: zap, a blazingly fast webframework written in Zig.

This project is heavily inspired by this post but we'll code it in Zig, a system programming language capable of handling thousands of requests (the original one was
written in Python). The Zig version used is 0.14 and the redis version is 7.0.15. In this blogpost there's only a few snippets of the code since most of it is not that important but you can check the full code here and this post was originally posted here.

OBS: since Redis is also a database, I'll use the term "database" referring to the database where the data is stored (in our case a hashmap) and use the term "redis" refering to the Redis database.

Implementation

To simulate a database, we'll use a hashmap instead of deploying one. Note that our database has different users indexed by an ID, which we'll be used to fetch them. The User struct is quite simple: has a id, a name, an email and an age.

var users = std.AutoHashMap(u32, entities.User).init(allocator);
defer users.deinit();

try users.put(1, entities.User{
    .id = 1,
    .name = "Alice",
    .email = "alice@example.com",
    .age = 25,
});
try users.put(2, entities.User{
    .id = 2,
    .name = "Bob",
    .email = "bob@example.com",
    .age = 30,
});
try users.put(3, entities.User{
    .id = 3,
    .name = "Charlie",
    .email = "charlie@example.com",
    .age = 22,
});

Make sure you install redis and that it's running:

sudo apt-get install redis-server
sudo systemctl start redis.service

First we need to connect to the redis running in the machine. To do it, we'll use okredis, a client for Redis written in Zig. If you're running it in Docker you gotta change the IP the same way I did in the repository.

pub fn main() !void {
    var client: Client = undefined;

    const addr = try net.Address.parseIp4("127.0.0.1", 6379);
    const connection = try net.tcpConnectToAddress(addr);

    try client.init(connection);
    defer client.close();

    const redischace: RedisCache = RedisCache{
        .ttl = "30",
        .client = client,
    };
}

The ideia is the following: we'll try to find the record in the cache. If it's not cached, we'll fetch them in the database and insert it into the cache. The implementation it's simple: first we need to try to get the value from redis:

switch (try self.redisCache.client.sendAlloc(
    OrErr([]u8),
    self.allocator,
    .{ "GET", id_request },
))

There's three possible cases from the switch case: if it's .Ok, we'll just use the value. If it's Nil (the ID is not in redis), fetch the database and insert it into redis. The last case is a .Err, where something goes wrong, and we'll just log the error by now.

{   .Ok => |value| {
        response = try self.allocator.dupe(u8, value);
    },
    .Nil => {
        const id_request_int = try std.fmt.parseInt(
            u8,
            id_request,
            10,
        );
        const id_request_casted: u32 = @intCast(id_request_int);

        if (self.users.getPtr(id_request_casted)) |user| {
            user.timestamp = std.time.timestamp();
            const json_string = try json.stringifyAlloc(
                self.allocator,
                user,
                .{},
            );
            defer self.allocator.free(json_string);

            try self.redisCache.client.send(void, .{ "SET", id_request, 
                                                    json_string, "EX", self.redisCache.ttl });
            response = try self.allocator.dupe(u8, json_string);
        } else {
            std.log.err("User {d} not found", .{id_request_casted});
        }
    },

    .Err => |err| std.log.err("error code = {any}\n", .{err.getCode()}),
}

Note that when we send to the client (in the .Nil case), we set a parameter "EX" (defined here in self.redisCache.ttl). The "EX" is the expiring time in seconds, also called "time to live" (ttl). The value we are using here is 30, so during 30 seconds every time the response will be the same. It can be verified by the "timestamp" field in the json.

Testing

To test it, we'll make a GET request with the following body:

{
    "id": "2"
}

It will fetch it and return the record in the "value" field:

{
    "id": "2",
    "value": "{\"id\":2,\"name\":\"Bob\",\"email\":\"bob@example.com\",
                \"age\":30,\"timestamp\":1741712135}"
}

Since we set a TTL of 30 seconds, if the call it again the result would be the same. After 30 seconds, the value is removed from redis.

If we call an ID that is not in the database, this should be the return.

{
    "id": "4",
    "value": ""
}

I just included the most important parts of the code here because I don't want to tire my beloved readers but, as always, the full code is on this public repository.
And that's all folks. Keep on learning :D

Running sklearn models in Zig

Vinícyus Brasil — Wed, 05 Feb 2025 22:55:54 +0000

Introduction

Scikit-learn models became the industry standard for creating machine learning models.Although the library offers many conveniences, one becomes tied to Python APIs for serving such models. These frameworks (such as FastAPI, for example) have various scalability issues, as Anton demonstrates in this video. To overcome this limitation, one solution is to use Python's C library to run the models, while the rest of the API can be built using another framework.

In this blogpost we'll implement this strategy. We'll create an application that uses Python's C API to run the models with the Zap library handling the requests,
which is a Zig blazing fast microframework for web applications.
The full code to the post can be found in this repository and this post was originally posted on my blog.

Creating the model and the shared object

So let's say we got this simple script called train.py that creates a Logistic Regression, in which the model takes a numpy array
with two dimensions and returns the value 1 or 0. The model is saved in a pickle file called model.pkl.

import numpy as np
from sklearn.linear_model import LogisticRegression
import pickle

X = np.array([
    [1.2, 1.3],
    [2.2, 1.4],
    [0.9, 2.1],
    [-1.2, -2.3],
    [-1.3, -0.2],
    [-1.1, -2.8],
])

y = np.array([[1],[1],[1],[0],[0],[0]])

y = y.ravel()
model = LogisticRegression()

model.fit(X, y)

with open('model.pkl', 'wb') as handle:
    pickle.dump(model, handle)

The full library we are going to create just loads the pickle file and returns it in a list. It will attend by the name of libpredict
and it predict_sklearn.py module have the following code:

import numpy as np
import pickle

def predict(x):
    with open('./model.pkl', 'rb') as handle:
        b_1 = pickle.load(handle)

    return [float(b_1.predict(np.array(x).reshape(1, -1))[0])]

Quite simple, right? The next step is to compile it to a shared object, the kind of file that contains the code we just wrote. Theoretically this step is not needed but I think its way easier to work with the library this way.

To compile the library we can use Cython. The following setup.py file will translate our library to C and generate the shared object.

from setuptools import setup, Extension, find_packages
from Cython.Build import cythonize
import sysconfig


python_inc = sysconfig.get_path("include")
python_lib = sysconfig.get_config_var("LIBDIR")
python_version = sysconfig.get_config_var("VERSION")
python_shared = f"python{python_version}"

extensions = cythonize(
    Extension(
        name="libpredict",
        sources=["libpredict/predict_sklearn.py"],
        libraries=[python_shared],  
        library_dirs=[python_lib],  
        include_dirs=[python_inc], 
        extra_link_args=[
            "-Wl,-rpath," + python_lib,  
            "-Wl,--no-as-needed",  
        ],
    )
)

setup(
    name="libpredict",
    packages=find_packages(),
    ext_modules=extensions,
)

Note that both of the files generated, libpredict.so and model.pkl, need to be moved to the same folder where we'll build the zig executable.

Running Python code in Zig

To interact with the Python C API, we gotta load the header file that includes the Python functions. The main.zig file starts with:

const c = @cImport({
    @cInclude("Python.h");
});

The syntax to call Python is in this documentation. For exemple, to print a string, we first need to initialize the interpreter and use the function PyRun_SimpleString with the string.

    c.Py_Initialize();
    _ = c.PyRun_SimpleString("import sys; sys.path.append('.');");
    c.Py_Finalize();

The main.zig file will start this way. First, we initialize the interpreter. Then, load the module libpredict and extracts from it a pointer to the function predict from the predict_sklearn.py. To call the function we need to convert the zig array doubles2 to a Python List also using the C API and that's the reason for the function prepareList to exist. Finally, the function can be called using the converted list.

pub fn main() !void {
    c.Py_Initialize();

    _ = c.PyRun_SimpleString("import sys; sys.path.append('.');");

    const module = try loadMod();

    const func = try loadFunc(module);

    var doubles2 = [_]f64{ 1.2, 1.3 };

    const pList = try prepareList(doubles2[0..]);

    const result = evaluteResult(pList, func);

    std.debug.print("Test: {any}\n", .{result});

    _ = c.Py_DecRef(func);
    _ = c.Py_DecRef(module);
}

Note the loadMod() function appends the current path and from it import the libpredict.so file so the loadFunc function can link a pointer to the predict function.

pub fn loadMod() !*c.PyObject {
    const sys = c.PyImport_ImportModule("sys");
    const path = c.PyObject_GetAttrString(sys, "path");
    _ = c.PyList_Append(path, c.PyUnicode_FromString("."));

    const module = c.PyImport_ImportModule("libpredict");
    return module;
}

pub fn loadFunc(module: *c.PyObject) !*c.PyObject {
    const func = c.PyObject_GetAttrString(module, "predict");
    return func;
}

The prepareList function just creates a Python list with PyList_New and appends to it the values from the doubles arrayh.

pub fn prepareList(doubles: []f64) !*c.PyObject {
    const pList = c.PyList_New(0);

    for (doubles) |elem| {
        const pValue = c.PyFloat_FromDouble(elem);

        if (c.PyList_Append(pList, pValue) != 0) {
            c.PyErr_Print();
        }
        _ = c.Py_DecRef(pValue);
    }

    return pList;
}

The evaluteResult function iterates through all of the values of the Python function's response and the returns the last one. I've written the code
this way because maybe what you are interested is the return of the predict_proba() function of the sklearn model, so it's easier to adapt the code.

pub fn evaluteResult(pList: *c.PyObject, func: *c.PyObject) !f64 {
    const args = c.PyTuple_Pack(1, pList);
    const value = c.PyObject_CallObject(func, args);

    var result: f64 = undefined;
    if (value != null) {
        const list_size: usize = @intCast(c.PyList_Size(value));

        for (0..list_size) |i| {
            const index: isize = @intCast(i);
            const pItem = c.PyList_GetItem(value, index);
            const pValue = c.PyFloat_AsDouble(pItem);

            std.debug.print("Result = {d}\n", .{pValue});
            result = pValue;
        }

        _ = c.Py_DecRef(value);
    } else {
        c.PyErr_Print();
    }
    _ = c.Py_DecRef(args);
    return result;
}

Creating the API with Zap

I've chosen Zap because it's a blazing fast micro webframework that can handle a lot of requests simultaneously and also it's quite intuitive.

To create the API, we need a Router object to map the routes to the functions that handle the requests. Our only route here is the predict that receives a json payload with the fields parameter1 and parameter2 to send to the Python function. We'll also need a HttpListener object to listen to the requests on the port 3000. The struct PredictorPackage is where all of the handling of the request is made;

pub fn main() !void {

    ...

    var gpa = std.heap.GeneralPurposeAllocator(.{
        .thread_safe = true,
    }){};

    const allocator = gpa.allocator();

    var simpleRouter = zap.Router.init(allocator, .{
        .not_found = not_found,
    });
    defer simpleRouter.deinit();

    const doubles = try allocator.alloc(f64, 2);
    defer allocator.free(doubles);

    var somePackage = PredictorPackage.init(
        allocator,
        doubles[0..],
        func,
    );

    try simpleRouter.handle_func("/predict", &somePackage, &PredictorPackage.predictValue);

    var listener = zap.HttpListener.init(.{
        .port = 3000,
        .on_request = simpleRouter.on_request_handler(),
        .log = true,
        .max_clients = 100000,
    });
    try listener.listen();

    std.debug.print("Listening on 0.0.0.0:3000\n", .{});

    zap.start(.{
        .threads = 2,
        .workers = 1,
    });
}

Inside the PredictorPackage struct, the function predictValue captures the body of the request, parses it with std.json.Parsed and then
calls the Python function with it. Then, it stringify it and responds the request with a resultPrediction object.

pub const valueToPredict = struct {
    parameter1: f64,
    parameter2: f64,
};

pub const resultPrediction = struct {
    predictedClass: f64,
};

pub const PredictorPackage = struct {
    const Self = @This();

    allocator: Allocator,
    doubles: []f64,
    func: *c.PyObject,

    pub fn init(
        allocator: Allocator,
        doubles: []f64,
        func: *c.PyObject,
    ) Self {
        return .{
            .allocator = allocator,
            .doubles = doubles,
            .func = func,
        };
    }

    pub fn predictValue(self: *Self, req: zap.Request) void {
        std.log.warn("predict_value_requested", .{});

        if (std.mem.eql(u8, req.method.?, "POST")) {
            std.debug.print("Preparing to predict\n", .{});

            req.parseBody() catch |err| {
                std.log.err("Parse Body error: {any}. Expected if body is empty", .{err});
            };

            if (req.body) |body| {
                const maybe_user: ?std.json.Parsed(std.json.Value) = std.json.parseFromSlice(std.json.Value, self.allocator, body, .{}) catch null;

                if (maybe_user) |parsed_user| {
                    defer parsed_user.deinit();

                    if (parsed_user.value.object.get("parameter1")) |parameter1| {
                        self.doubles[0] = parameter1.float;
                    }
                    if (parsed_user.value.object.get("parameter2")) |parameter2| {
                        self.doubles[1] = parameter2.float;
                    }
                    std.log.info("Parameter1: {?}", .{self.doubles[0]});
                    std.log.info("Parameter2: {?}", .{self.doubles[1]});
                } else {
                    std.log.err("Failed to parse JSON", .{});
                }
            }

            // predict

            std.debug.print("vector to predict: {any}\n", .{self.doubles[0..]});

            const pList = try prepareList(self.doubles[0..]);

            const result = try evaluteResult(pList, self.func);

            std.debug.print("resultado: {any}\n", .{result});

            //transform prediction into json
            const resultSend = resultPrediction{ .predictedClass = result };

            var buf: [100]u8 = undefined;
            var json_to_send: []const u8 = undefined;
            if (zap.stringifyBuf(&buf, resultSend, .{})) |json| {
                json_to_send = json;
            } else {
                json_to_send = "null";
            }

            // send prediction back to the client
            req.setContentType(.JSON) catch return;
            req.sendBody(json_to_send) catch return;
        }
    }
};

The full main.zig file, as well as a dockerfile with all of the components, can be found in the repository of the project.

After building the project (zig build run) and moving the libpredict.so and the model.pkl to the same folder as the executable,
we can test the API with cURL:

curl -X POST http://localhost:3000/predict \
     -H "Content-Type: application/json" \
     --data '{"parameter1": -1.3, "parameter2": -1.4}'

It should return the following response:

{
    "predictedClass": 0e0
}

And that's it. Fell free to contact me via Linkedin or to open a PR on the GitHub repo if you find something wrong. Keep on learning :D

Building a real-time Websocket webapp with Kafka and Javascript

Vinícyus Brasil — Fri, 24 Jan 2025 23:17:24 +0000

Introduction

Suppose we have a API that generates a lot of information whom we must store in a database and show them to clients. This information updates every few seconds
and this sort of system is call a real-time app (in reality, it takes some seconds, so it's a near real time app). Some applications that can use this sort of design are:

chatbots: the API is a LLM that receives an input and send it's response to the client;
systems with time series: the API fetches data from somewhere (like some temperature reading) and updates in real time to the client;
systems with position in space: the API fetches data from a train (or a car) and updates show them to someone waiting (like a Uber passenger);
sports betting: the API calculate the odds of something happen in the game and updates it to the client.

In this project we'll build a near real-time app. We've chosen to build a simple sports betting app that will only show the clients updated odds.
The webapp will send a websocket message to their clients after
receiving a message from a Kafka topic. This message comes from a new record being inserted in the database
from a API and captured by Kafka-Connect with the debezium plugin.

The design of the system follows: a API receives external data and insert it to the database, Kafka Connect get the payload inserted and send it to
a Kafka topic. A websocket server is consuming the messages from this topic and, when receives a message, send it to the websockets clients.
This design allows the system to be scalable since some of the components can work asynchronously, therefore the ones with more load can be horizontally scaled.
The image below ilustrates it.

There are four main steps to build this system:

1) setup Kafka and Kafka connect with the debezium plugin;
2) setup the database (MySQL) and the scripts to initialize it correctly;
3) build the API to create the data;
4) create the websocket server and the client;

Here I'll only share some parts of the code because the full project is too large and the full code is on
this Github repository and this was originally posted on my blog.

Setting up Kafka Connector and debezium (the docker-compose file)

To use Kafka we need Zookeeper, also included in the following docker-compose file. The file also creates the database and uses default password and username.
Kafka-Connect installing the debezium plugin is where the magic happens. The plugin is responsable for get the data inserted in the database and push it to a
Kafka topic. The version of docker used is 20.10.12, the docker-compose version is v2.29.1-desktop.1 and here's the code for the docker-compose file.

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.8.0
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.8.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 100
    depends_on:
      - zookeeper

  mysql:
    image: mysql:8.0
    container_name: mysql
    ports:
      - 3306:3306
    environment:
     - MYSQL_ROOT_PASSWORD=debezium
     - MYSQL_USER=mysqluser
     - MYSQL_PASSWORD=mysqlpw
    volumes:
     - ${PWD}/data/mysql:/docker-entrypoint-initdb.d
     - ${PWD}/data:/data
     - ./mysql-init:/docker-entrypoint-initdb.d
  schema-registry:
    image: confluentinc/cp-schema-registry:7.8.0
    container_name: schema-registry
    ports:
      - "8081:8081"
    depends_on:
      - kafka
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:29092
  kafka-connect:
    image: confluentinc/cp-kafka-connect-base:7.8.0
    container_name: kafka-connect
    depends_on:
      - kafka
      - schema-registry
    ports:
      - 8083:8083
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka:29092"
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: kafka-connect
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
      CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"
      CONNECT_LOG4J_APPENDER_STDOUT_LAYOUT_CONVERSIONPATTERN: "[%d] %p %X{connector.context}%m (%c:%L)%n"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components,/data/connect-jars
    volumes:
      - $PWD/data:/data
    command:
      - bash
      - -c
      - |
        echo "Installing Connector"
        confluent-hub install --no-prompt debezium/debezium-connector-mysql:1.7.0
        #
        echo "Launching Kafka Connect worker"
        /etc/confluent/docker/run &
        #
        sleep infinity

networks:
  my-network-name:
   name: websockets

Note we are naming the network that docker will use. It is importante because all of the services must be in the same network. To run it:

docker-compose up -f docker-compose.yml -d

Setting up the database

Each game will use a table in the database "gameodds" and the table will be called game_{gameid}, where {gameid} is a int
identifing each game. By creating the database we also need to grant the privileges to the user (please, don't let
credentials like this explicity in the code in production, you're gonna get yourself fired).

I chose MySQL because it's simple but you can use another database since debezium has plugins in their website for MongoDB, PostgreSQL etc.
The following script file will to be in the 'mysql-init' folder and it will run every time docker-compose up.

CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz';

GRANT ALL PRIVILEGES ON *.* TO 'debezium'@'%' WITH GRANT OPTION;

FLUSH PRIVILEGES;


CREATE DATABASE IF NOT EXISTS gameodds;

USE gameodds;

CREATE TABLE IF NOT EXISTS available_games (
    row_id INT AUTO_INCREMENT PRIMARY KEY,
    game_id VARCHAR(50) NOT NULL,
    game_id_name VARCHAR(50) NOT NULL,
    connector_name VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

API to send request to database

The API needs two routes:

/games: the game odds will be stored in a table called "available_games". Calling a POST request to this route will create the table and then
send a POST request to Kafka-Connect to create the connector,
which will create the topic in Kafka and send the payloads;
/sendodd: this route generates a random payload and inserts it into the database (in a real scenario, it would calculate the odd based on some
external data);

A simple payload will be sent to the database with some random values:

{
    "game_id": "123121",
    "odd1": 1.07,
    "odd2": 3.54,
    "odd3": 0.96,
    "odd4": 3.34,
    "odd5": 1.96
}

The simpler way to create the API for this project it's by using FastAPI. When the program starts, it will run a function called "startup_event"
and connect the program to the database.

import mysql.connector 
import random
import requests

from fastapi import FastAPI

app = FastAPI()


KAFKA_CONNECT_IP = "172.18.0.6"
MYSQL_IP = "172.18.0.2"

@app.on_event("startup")
async def startup_event():
    global mydb
    mydb = mysql.connector.connect(
    host=MYSQL_IP,
    user="debezium",
    password="dbz",
    database="gameodds"
    )

Note that two variables are used: KAFKA_CONNECT_IP and MYSQL_IP, both internal IPs of the containers. To get them, we can inspect manually using docker network inspect websockets_default or with this:

docker inspect --format='{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $(docker ps -aq --filter "name=mysql")

docker inspect --format='{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $(docker ps -aq --filter "name=kafka-connect")

The /games route creates a new table using pure SQL based on the {game_id} from the request. Then, it calls the kafka-connect API creating a new connector with
this table.

@app.post("/games/{game_id}")
def create_game(game_id: str):
    mycursor = mydb.cursor()

    sql = f'''
        CREATE TABLE IF NOT EXISTS game_{game_id} (
            row_id INT AUTO_INCREMENT PRIMARY KEY,
            game_id VARCHAR(50) NOT NULL,
            odd1 VARCHAR(50),
            odd2 VARCHAR(50),
            odd3 VARCHAR(50),
            odd4 VARCHAR(50),
            odd5 VARCHAR(50),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );   
    '''

    mycursor.execute(sql)

    mydb.commit()

    mycursor = mydb.cursor()

    sql = f'''INSERT INTO available_games (game_id, game_id_name, connector_name) 
        SELECT '{game_id}', '{"game_" + str(game_id)}', 
        '{"mysql-debezium-json-no-schema-asgard.gameodds.game_" + str(game_id)}' 
        FROM DUAL WHERE NOT EXISTS ( SELECT 1 FROM available_games WHERE game_id = '{game_id}' ); 
        '''
    print(sql)
    mycursor.execute(sql)

    mydb.commit()

    create_connector = requests.post(f"http://{KAFKA_CONNECT_IP}:8083/connectors/", headers={"Accept": "application/json", "Content-Type":"application/json"}, json={
      "name": f"source-debezium-orders-{game_id}",
      "config": {
            "connector.class": "io.debezium.connector.mysql.MySqlConnector",
            "database.hostname": "mysql",
            "database.port": "3306",
            "database.user": "debezium",
            "database.password": "dbz",
            "database.server.id": f"{game_id}",
            "database.server.name": "asgard",
            "table.whitelist": f"gameodds.game_{game_id}",
            "database.history.kafka.bootstrap.servers": "kafka:29092",
            "database.history.kafka.topic": f"dbserver_{game_id}" ,
            "decimal.handling.mode": "double",
            "include.schema.changes": "true",
            "value.converter": "org.apache.kafka.connect.json.JsonConverter",
            "value.converter.schemas.enable": "false",
            "key.converter": "org.apache.kafka.connect.json.JsonConverter",
            "key.converter.schemas.enable": "false",
            "transforms": "unwrap,addTopicPrefix",
            "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
            "transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
            "transforms.addTopicPrefix.regex":"(.*)",
            "transforms.addTopicPrefix.replacement":"mysql-debezium-json-no-schema-$1",
            "database.allowPublicKeyRetrieval": "true"
       }
    })

The /sendodd route just creates the odds randomly and insert it into the table.

@app.post("/sendodd/{game_id}")
def read_root(game_id: str):
    mycursor = mydb.cursor()

    odd1 = round(random.random() * 5, 2)
    odd2 = round(random.random() * 5, 2)
    odd3 = round(random.random() * 5, 2)
    odd4 = round(random.random() * 5, 2)
    odd5 = round(random.random() * 5, 2)

    sql = f'INSERT INTO game_{game_id} (game_id, odd1, odd2, odd3, odd4, odd5) VALUES ({game_id}, {str(odd1)}, {str(odd2)}, {str(odd3)}, {str(odd4)}, {str(odd5)})'

    mycursor.execute(sql)

    mydb.commit()

    print(mycursor.rowcount, "record inserted.")

    return {"game_id": game_id, "odd1": odd1, "odd2": odd2, "odd3": odd3, "odd4": odd4, "odd5" : odd5}

Containerazing it with Docker (check the Dockerfile in the repository) and building the project (note we are using the "net" argument so it can access the database ):

docker build -t fastapi-image . && docker run -d --name fastapi-container -p 80:80 --net websockets_default fastapi-image

So to create a game with the id = 123122:

curl -X POST "localhost:80/games/123122"

Send some odds in that game to the database:

curl -X POST "localhost:80/sendodd/123122"

JS Websocket server

The first thing we'll need is the websocket server, which will receive the messages from the Kafka topic and send them
to all the websocket clients. The version of the package used is ws@8.18.0. The following code uses Javascript to create a websocket server
and the consumer to the Kafka topics.

Let's load the libraries needed and start the connection with the database, start the connection with Kafka and start the Websocket server:

const WebSocket = require("ws");
var mysql = require("mysql2");
const { Kafka } = require("kafkajs");

const MYSQL_IP = "172.18.0.2";
const KAFKA_IP = "172.18.0.5";

let connection_names = [];

var con = mysql.createConnection({
  host: MYSQL_IP,
  port: "3306",
  user: "debezium",
  password: "dbz",
  database: "gameodds",
});

const wss = new WebSocket.Server({ port: 81 });

console.log("Server running");


const kafka = new Kafka({
  brokers: [`${KAFKA_IP}:29092`],
});

const consumer = kafka.consumer({ groupId: "group-1" });

con.connect(function (err) {
  if (err) throw err;
  con.query(
    "SELECT connector_name FROM available_games",
    function (err, result, fields) {
      if (err) throw err;
      var values = result.map((item) => item.connector_name);
      connection_names = values;
      run(connection_names).catch(console.error);
    }
  );
});

Every time the server starts, it queries the database looking for available games. They are stored in a list
called "connection_names" and the server will consume the Kafka topic related to that game. Every time
it receives a new payload from any of the games, it stores in a variable call last_odds. Them, every 5 seconds, it sends it to all the clients connected.

var last_odds = {};

const run = async (connection_names) => {
  await consumer.connect();


  await connection_names.forEach((item) => {
    console.log(`connecting: ${item}`);
    consumer.subscribe({
      topics: [item],
      fromBeginning: false,
    });
  });


  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        value: message.value.toString(),
        topic,
        partition,
      });
      last_odds = JSON.parse(message.value.toString());
    },
  });
};

wss.on("connection", (ws) => {
  console.log("Client connected");

  console.log("Received a new connection");

  ws.on("close", () => {
    console.log("Client disconnected");
  });
});

function update() {
  console.log(last_odds);
  console.log(connection_names);

  wss.clients.forEach((client) => {
    if (client.readyState === WebSocket.OPEN) {
      client.send(JSON.stringify(last_odds));
    }
  });
}

const intervalId = setInterval(update, 5000);

OBS: In the utils/ folder of the project there's a js file with a test. It will just open a connection, send a websocket message to the server
and then close it. It's just for testing pourposes but it can be useful if something isn't working.

JS Websocket client

For the client, we'll use React to build a simple UI. The version of the packages used are create-react-app@5.0.1, web-vitals@4.2.4 and ws@8.18.0.
It uses a list of games indexed by the game_id and shows them. If it receives a message with the same game_id of one of the messages stored in the list,
it updates the values of the odds. If the game_id is not among them, just appends it to the list. The code also connects itself to the websocket server we've build
earlier.
Creating the app with create-react-app, the App.js will be:

import "./App.css";

import React, { useEffect, useState } from "react";

const WEBSOCKETSERVER_IP = "127.0.0.1";
const socket = new WebSocket(`ws://${WEBSOCKETSERVER_IP}:81`);

const OutputComponent = ({ messages }) => {
  return (
    <div id="output">
      {messages.map((game, index) => (
        <div key={index} className="container">
          <ul className="list">
            <li>
              <div>Game ID: {game.game_id}</div>
              <div>Odd 1: {game.odd1}</div>
              <div>Odd 2: {game.odd2}</div>
              <div>Odd 3: {game.odd3}</div>
              <div>Odd 4: {game.odd4}</div>
              <div>Odd 5: {game.odd5}</div>
            </li>
          </ul>
        </div>
      ))}
    </div>
  );
};




function App() {
  const [messages, setMessages] = useState([]);

  useEffect(() => {
    socket.onopen = function (event) {
      console.log("You are Connected to WebSocket Server");
    };



    socket.onmessage = function (event) {
      const payload = JSON.parse(event.data);

      setMessages((prevMessages) => {
        const existingIndex = prevMessages.findIndex(
          (msg) => msg.game_id === payload.game_id
        );

        if (existingIndex >= 0) {
          const updatedMessages = [...prevMessages];
          updatedMessages[existingIndex] = {
            ...updatedMessages[existingIndex],
            ...payload,
          };
          return updatedMessages;
        } else {
          return [...prevMessages, payload];
        }
      });
    };

    socket.onclose = function (event) {
      console.log("Disconnected from WebSocket server");
    };
  });

  return (
    <div className="App">
      <header className="App-header">
        <h1>Odds Market Simulator</h1>

        <OutputComponent messages={messages} />
      </header>
    </div>
  );
}

export default App;

The Dockerfile for the WebSocket client uses nginx to serve the built app. The final app looks something like this:

Testing

If everything is ok, by starting the docker-compose, the API, the websocker server and the client, we can now call the API to create a new game:

curl -X POST "localhost:80/games/123121"

Now lets create some odds:

curl -X POST "localhost:80/sendodd/123121"

It returns the following payload:

{
    "game_id": "123121",
    "odd1": 0.32,
    "odd2": 2.28,
    "odd3": 3.14,
    "odd4": 0.99,
    "odd5": 0.17
}

We should see the odds updated in the browser:

By calling it again, the odds should change:

curl -X POST "localhost:80/sendodd/123121"

It returns the following payload:

{
    "game_id": "123121",
    "odd1": 3.54,
    "odd2": 1.07,
    "odd3": 2.12,
    "odd4": 4.42,
    "odd5": 3.36
}

and updates in the browser:

Let's create another game now.

curl -X POST "localhost:80/games/123120"

By calling the endpoint, the new game should apper in the browser.

That's it. You can check the full source code in the repository and feel free to
contact me if there are some errors. Still there's a lot to be done and in the repository listed some thing in the todo list. Keep on learning :D

Let's create another game now.

curl -X POST "localhost:80/games/123120"

By calling the endpoint, the new game should apper in the browser.