SSE events

app.py

import sys
import queue
import threading
import time
import datetime
import uuid
from concurrent.futures import ThreadPoolExecutor

import flask

app = flask.Flask(__name__)
executor = ThreadPoolExecutor(max_workers=3)
qdict = dict()

def format_sse(data, event=None):
    if event is not None:
        msg = 'event: %s\ndata: %s\n\n' % (event, data)
    else:
        msg = 'data: %s\n\n' % data
    return msg

@app.route("/eventstream/<eventstream>", methods=["GET"])
def sse_method(eventstream):
    print('we are in eventstream handler', file=sys.stderr)
    q = qdict[eventstream]
    def stream():
        while True:
            print('fetching item from q', file=sys.stderr)
            item = q.get()
            print('got item: %s' % item, file=sys.stderr)
            if item is None:
                q.task_done()
                break
            yield format_sse('[%s]: %s' % (datetime.datetime.now().isoformat(), item))
            q.task_done()
        #flask.g.pop(eventstream, None)
        qdict.pop(eventstream)

    return flask.Response(stream(), mimetype='text/event-stream')


@app.route("/", methods=["GET"])
def root_method():
    q = queue.Queue(10)
    
    def job():
        #print('started', file=sys.stderr)
        q.put_nowait('started')
        #print('sleeping for 2 seconds', file=sys.stderr)
        time.sleep(2)
    
        #print('almost done (50%)', file=sys.stderr)
        q.put_nowait('almost done (50%)')
        time.sleep(2)
    
        #print('99%', file=sys.stderr)
        q.put_nowait('99%')
        time.sleep(2)
        print('done with sleep', file=sys.stderr)
    
        #print('100% - done!', file=sys.stderr)
        q.put_nowait('100% - done!')
        q.put_nowait(None)
        q.join()
        #print('consumer done')

    eventstream = str(uuid.uuid4())
    threading.Thread(target=job).start()
    qdict[eventstream] = q
    temp_url = "http://localhost:5000/eventstream/" + eventstream
    return flask.redirect(temp_url, 303)