Websockets

[josejachuf]

Hi @gi0baro

What would be missing in emmett to be able to do this [1]. Websockets and something else?

[1] https://programadorwebvalencia.com/tutorial-flask-para-crear-chat-con-socketio-y-vuejs/

[gi0baro]

Hi @josejachuf,
with current alpha (2.0.0a3) everything is already in place to use websockets, just the documentation is missing.

Mind that I never used socketio, but inspecting the snippets in the link you provided, the websocket implementation is a bit low level, so to implement a chat I'll probably use an external pub-sub provider like redis, and bind the sockets to it in order to push new messages.

[josejachuf]

@gi0baro
Can you show any minimal example of how to use it until the documentation is ready?
Thanks

[gi0baro]

@josejachuf sorry for the delay, I was a little busy in the last days.

BTW, here is how I implemented it, but please consider that sockets get used only to get new messages in real time, since for posting new messages I think a standard POST request is better. Also, this is implemented using redis since it saves quite a lot of coding.

First, you need an async implementation of redis (based on aioredis package):

from aioredis import Redis as _Redis
from aioredis.pool import ConnectionsPool
from contextlib import asynccontextmanager
from emmett.extensions import Extension, listen_signal

class Redis(Extension):
    def on_load(self):
        self.pool = None

    @listen_signal('after_loop')
    def build_pool(self, loop):
        self.pool = ConnectionsPool(
            f'redis://{self.app.config.redis.host}:{self.app.config.redis.port}',
            db=0,
            minsize=0,
            maxsize=128,
            create_connection_timeout=1,
            loop=loop
        )

    @asynccontextmanager
    async def ctx(self):
        connection = await self.pool.acquire()
        try:
            yield _Redis(connection)
        finally:
            self.pool.release(connection)

put the redis config in app.config.redis and use the extension:

app.use_extension(RedisExtension)

and then you can write your socket:

@app.websocket('/chat/<str:room>')
async def chat_messages(room):
    async with app.ext.Redis.ctx() as redis:
        channel, = await redis.subscribe(f'chat:{room}:messages')
        await websocket.accept()

        async for message in channel.iter():
            await websocket.send(message)

Now the only thing you need to do is to write the message to the redis channel after saving the record in your standard route, eg:

from emmett.serializers import Serializers
json_dump = Serializers.get_for('json')

@app.route('/chat/<str:room>/messages', methods=['post'])
async def new_chat_message(room):
    response.status = 201
    params = await request.body_params
    # maybe do some parsing?
    res = ChatMessage.create(**params)
    if res.errors:
        response.status = 422
        return {'error': res.errors.as_dict()}
    row = ChatMessage.get(res.id)
    async with app.ext.Redis.ctx() as redis:
        await redis.publish(f'chat:{room}:messages', json_dump(row))
    return row

It should work. Please mind that sockets shares the same pipeline of the routes, so if you put database pipe in the application pipeline it will open up connections also for every socket connection. My suggestion is to split pipelines with modules, something like:

routes = app.module(__name__, 'routes')
sockets = app.module(__name__, 'sockets')

routes.pipeline = [db.pipe]

I think is a good case for creating a proper example in the examples folder.
Let me know.

[josejachuf]

Thanks @gi0baro
I'm going to try it on the weekend

[josejachuf]

Hi @gi0baro

This ws is very new to me, I am trying to understand. When I try with (plugin ws client tester) here I get an error

@app.websocket('/chat/<str:room>')
async def chat_messages(room):
    async with app.ext.Redis.ctx() as redis:
        channel, = await redis.subscribe(f'chat:{room}:messages')
        await websocket.accept()

        async for message in channel.iter():
            await websocket.send(message)

As expected I get:

await websocket.accept()
NameError: name 'websocket' is not defined

[gi0baro]

@josejachuf

from emmett import websocket

[josejachuf]

Hi @gi0baro

import random
import string

from emmett import App, response, request, websocket
from emmett.serializers import Serializers
json_dump = Serializers.get_for('json')
from emmett_redis import RedisExtension

app = App(__name__)
# app.config.RedisExtension.host = "192.168.1.45"
app.config.RedisExtension.host = "127.0.0.1"
app.config.RedisExtension.port = '6379'
app.config.RedisExtension.password = "foobared"
app.use_extension(RedisExtension)


@app.websocket('/chat')
async def chat_messages():
    async with app.ext.RedisExtension.ctx() as redis:
        channel, = await redis.subscribe('chat:message')
        await websocket.accept()

        async for message in channel.iter():
            await websocket.send(message)

@app.route('/chat/messages')
async def new_chat_message():
    message = ''.join([random.choice(string.ascii_letters + string.digits) 
                         for n in range(32)])
    async with app.ext.RedisExtension.ctx() as redis:
        await redis.publish('chat:message', message)
    return message

I am quite lost. This is a simplification of your example. It works without errors.
The question I have, even more after having seen all the examples with flask and express, is how does the javascript client connect to the socket? I tried with socket.io

[josejachuf]

I got to that:

<!DOCTYPE html>
<html>
    <head>
        <title>WebSocket Emmett</title>
    </head>
    <body>
        <script>
            var socket = new WebSocket('ws://' + document.domain + ':' + location.port + '/chat');
            var messages = document.createElement('ul');
            // Escucha por mensajes
            socket.addEventListener('message', function (event) {
                reader = new FileReader();
                reader.readAsText(event.data);

                reader.onload = () => {
                    messages = document.getElementsByTagName('ul')[0];
                    var message = document.createElement('li');
                    var content = document.createTextNode('Received: ' + reader.result);
                    message.appendChild(content);
                    messages.appendChild(message);
                };
            });

            document.body.appendChild(messages);
        </script>
    </body>
</html>

[josejachuf]

Hello
With the change of aioredis from version 1.3.0 to the current 2.0.0 this has been broken and can not work.

The modifications I made were these

class RedisExtension(Extension):
    def on_load(self):
        self.pool = None

    @listen_signal(Signals.after_loop)
    def build_pool(self, loop):
        self.pool = aioredis.ConnectionPool.from_url("redis://10.1.1.42", max_connections=10)
       
    @asynccontextmanager
    async def ctx(self):
        try:            
            yield aioredis.Redis(connection_pool=self.pool)
        finally:
            pass
            # self.pool.release(connection)

@app.websocket('/chat/<str:room>')
async def chat_messages(room):
    async with app.ext.RedisExtension.ctx() as redis:
        ps = redis.pubsub()       
        channel, = await ps.subscribe(f'chat:{room}:messages')              
        await websocket.accept()
        async for message in channel.iter():
            await websocket.send(message)

I get this error:

channel, = await ps.subscribe(f'chat:{room}:messages')
TypeError: cannot unpack non-iterable NoneType object