Why do I need redis?

[vidstige]

This should be called flask_redis_sse instead.

[singingwolfboy]

@vidstige Do you have a suggestion for how to implement server-sent events without Redis? If so, please let me know. I'm happy to make this module pluggable, if you want to write a different backend.

[vidstige]

Yeah, I'm currently using the following

from typing import Iterator
import random
import string

from collections import deque
from flask import Response, request
from gevent.queue import Queue
import gevent

def generate_id(size=6, chars=string.ascii_lowercase + string.digits):
    return ''.join(random.choice(chars) for _ in range(size))

class ServerSentEvent(object):
    """Class to handle server-sent events."""
    def __init__(self, data, event):
        self.data = data
        self.event = event
        self.event_id = generate_id()
        self.desc_map = {
            self.data: "data",
            self.event: "event",
            self.event_id: "id"
        }

    def encode(self) -> str:
        """Encodes events as a string."""
        if not self.data:
            return ""
        lines = ["{}: {}".format(name, key)
                 for key, name in self.desc_map.items() if key]

        return "{}\n\n".format("\n".join(lines))

class Channel(object):
    def __init__(self, history_size=32):
        self.subscriptions = []
        self.history = deque(maxlen=history_size)
        self.history.append(ServerSentEvent('start_of_history', None))

    def notify(self, message):
        """Notify all subscribers with message."""
        for sub in self.subscriptions[:]:
            sub.put(message)

    def event_generator(self, last_id) -> Iterator[ServerSentEvent]:
        """Yields encoded ServerSentEvents."""
        q = Queue()
        self._add_history(q, last_id)
        self.subscriptions.append(q)
        try:
            while True:
                yield q.get()
        except GeneratorExit:
            self.subscriptions.remove(q)

    def subscribe(self):
        def gen(last_id) -> Iterator[str]:
            for sse in self.event_generator(last_id):
                yield sse.encode()
        return Response(
            gen(request.headers.get('Last-Event-ID')),
            mimetype="text/event-stream")

    def _add_history(self, q, last_id):
        add = False
        for sse in self.history:
            if add:
                q.put(sse)
            if sse.event_id == last_id:
                add = True

    def publish(self, message):
        sse = ServerSentEvent(str(message), None)
        self.history.append(sse)
        gevent.spawn(self.notify, sse)

    def get_last_id(self) -> str:
        return self.history[-1].event_id

And then I can just use it flask like so

import flask_sse
channel = flask_sse.Channel()

@app.route('/subscribe')
def subscribe():
    return channel.subscribe()

@app.route('/publish')
def publish():
    channel.publish('message here')
    return "OK"

This works very smooth and does not require Redis. Perhaps can the gevent specific parts be moved out somehow? This can be combined with gunicorn or the gevent built-in WSGI for example.

[singingwolfboy]

Nifty! So, we could have at least two backends: a Redis backend, and a gevent backend. Maybe it's possible to have an asyncio backend, as well!

I don't have time to rewrite this project to be pluggable, but if you'd like to submit a pull request, I'd be happy to review it and hopefully merge it! I'd also be happy to find a time to chat with you further (over text or videochat) to discuss how best to structure the code to make this happen.

[vidstige]

Alright, cool! Thanks for your feedback. I also have very little time normally, but every once in a while I enter a weekend hacking-spree. Will try to checkout your code and see if it would be possible to pass in different backend somehow. I guess even with Redis you need some kind of concurency even in the webserver for it not to block while streaming the SSE... :thinking:

[rlam3]

would tornado help with the async process?

[vidstige]

Yeah. I prefer gunicorn / gevent though.

[georgreen]

Am trying to learn and do some open source projects. Am currently using this and the redis part is a real blocker because most of the time we have to pay for the redis server resource. I would like to take part in this project. @vidstige @singingwolfboy

[vidstige]

One way would be to add my gevent stuff above, and then rename the redis stuff so you would name the backend you want. Then, as a step two we will see what code is common, and then this code could be refactored out. As an optinal step 3, the backend options could be passed in as a class or so. Voila - A sweet pluggable flask-sse module.

[georgreen]

@vidstige I have your code offline I have been testing it out though I hit a snag on the publish, I can't seem to get the publish to send some data plus am not sure how to create different channels which the clients can subscribe to. Any insight on how to go about this?

[vidstige]

@georgreen it sounds like you're not running your flask app in a gevent container. You need to use a gevent container such as gnicorn. Easiest to get started is to use the built-in gevent.wsgi module, like so

from flask import Flask, request
from gevent.wsgi import WSGIServer
import flask_sse

app = Flask(__name__)
channel = flask_sse.Channel()

@app.route('/subscribe')
def subscribe():
    return channel.subscribe()

@app.route('/publish', methods=['POST'])
def publish():
    channel.publish(request.data)
    return "OK"

@app.route('/')
def index():
    return """<body><script>
var eventSource = new EventSource('/subscribe');
eventSource.onmessage = function(m) {
    console.log(m);
    var el = document.getElementById('messages');
    el.innerHTML += m.data;
    el.innerHTML += "</br>";
}
function post(url, data) {
    var request = new XMLHttpRequest();
    request.open('POST', url, true);
    request.setRequestHeader('Content-Type', 'text/plain; charset=UTF-8');
    request.send(data);
}
function publish() {
    var message = document.getElementById("msg").value;
    post('/publish', message);
}
</script>
<input type="text" id="msg">
<button onclick="publish()">send</button>
<p id="messages"></p>
</body>"""

def main():
    server = WSGIServer(("", 5000), app)
    server.serve_forever()

if __name__ == "__main__":
    main()

This app you should be able to run directly. :-) The "flask_sse" is the code I pasted above. Good luck.

[vidstige]

The thing is, if you do not run a gevent container, the subscribe call will block everything and the publish code will never be called.

[georgreen]

@singingwolfboy I was trying run notifications in a background process where I call the publish method several times, but surprisingly this does not work. It fails silently, I have been unable to pin down the problem. Can you please point me in the right direction on how I can resolve this?

[georgreen]

@vidstige Redis is essential in the case where you need to run several instances of the app, and you need all of them to be in sync. e.g When using gunicorn with more than one worker, using non-redis implementation result into inconsistent behavior.

[singingwolfboy]

As I said, I don't have the time (or the interest) to rewrite this project myself to be pluggable with several different backends, and I certainly can't answer questions or provide support for other backends. Redis seems to work well for this, but I have no idea how well gunicorn works.

However, if someone else wants to step up and write the code to make this pluggable, as well as writing a gunicorn backend, that would be great! I'd be happy to review the code and documentation. I'm also happy to have a high-level discussion about how to make this change, before any code is written.

[vidstige]

@georgreen yeah, thats true, if you have several workers redis is probably a good idea. If you have a larger site, you are likely to have redis anyway.

[singingwolfboy]

Since there's been no discussion on this topic for several years, I'm closing this issue.

[Kamforka]

Just a late question. If you run multiple instances of your SSE app (e.g. load balanced behind a reverse proxy), then the in-process queue solution will eventually lead to out of sync event subscriptions, therefore a Redis back-end is superior, is it correct?

[vidstige]

Yes, if you have multiple processes, you will need redis or something similar (RabbitMQ?). The in-process way is only for situations with a single process. So both ways are "superior" depending on the situation.

[vsnthdev]

For anybody still looking for a solution. I've published a Queue based implementation (mainly for ease of use) 👇 https://github.com/vsnthdev/flask-queue-sse