Stirfried 🥡

Socket.IO server to schedule Celery tasks from clients in real-time.

Getting started

Stirfried has a three layered architecture:

1. Socket.IO clients
2. Socket.IO server
3. Celery workers

The design allows you to independently scale the number of servers when server-client communication workload increases and the number of workers when the task processing workload increases.

By leveraging Celery's task routing (explained below) you can also divide workers into groups and scale groups independently.

Socket.IO clients

Clients can connect using standard Socket.IO libraries.

The server is listening for clients to emit any of the following events:

Clients can subscribe to the following events emitted by the server:

Socket.IO server

For the Socket.IO server component you can pull the prebuilt docker image:

docker pull korijn/stirfried

or you can copy the project and customize it to your liking.

You are required to provide a settings.py file with the configuration for the server.

It requires at a minimum:

• socketio_redis - Redis connection string for the Socket.IO server.
• broker_url - Connection string for the Celery broker.

Task routing

The server sends tasks to the Celery broker by name, so it can act as a gateway to many different Celery workers with different tasks. You can leverage Celery's task routing configuration for this purpose.

Example

Let's say you have two workers, one listening on the feeds queue and another on the web queue. This is how you would configure the server accordingly with settings.py:

socketio_redis = "redis://localhost:6379/0"
broker_url = "redis://localhost:6379/1"
task_routes = {
    "feed.tasks.*": {"queue": "feeds"},
    "web.tasks.*": {"queue": "web"},
}

You can then run the server as follows:

docker run --rm -ti -v `pwd`/settings.py:/app/settings.py:ro -p 8000:8000 stirfried

Celery workers

You need to install Stirfried in your Celery workers via pip:

pip install stirfried

In your Celery workers, import the StirfriedTask:

from stirfried.celery import StirfriedTask

Configure StirfriedTask as the base class globally:

app = Celery(..., task_cls=StirfriedTask)

...or per task:

@app.task(base=StirfriedTask)
def add(x, y, room=None):
    return x + y

Rooms

The server injects the client's sid into the keyword argument room.

The StirfriedTask base class depends on the presence of this keyword argument.

This means you are required to add the keyword argument room=None to your task definitions in order to receive it.

Progress

You can emit progress from tasks by calling self.emit_progress(current, total).

Note that you are required to pass bind=True to the celery.task decorator in order to get access to the self instance variable.

@celery.task(bind=True)
def add(self, x, y, room=None):
    s = x
    self.emit_progress(50, 100)  # 50%
    s += y
    self.emit_progress(100, 100)  # 100%
    return s