HTTP Server Push - ارسال داده ها به صورت استریم در پروتکل SSE با پایتون

HTTP Server Push یا HTTP Streaming یک ارتباط بین سرور و کاربر هست که فقط کافیه کاربر اتصال رو برقرار کنه بعد از اون داده ها بدون اینکه کاربر درخواست بزنه سمتش ارسال میشه.

چند تا پروتکل برای ارسال داده ها به صورت استریم وجود داره.

HTTP Long Pulling یا Hanging GET

در Long Pulling وقتی کاربر درخواستش رو ارسال میکنه، سرور چک میکنه که داده جدید برای ارسال هست یا نه، اگه باشه که ارسال میکنه و اتصال بسته میشه مثل یک درخواست معمولی ولی اگه داده جدیدی نباشه چیزی برنمیگردونه و اتصال رو باز نگه میداره تا داده جدیدی آماده بشه و وقتی داده جدید آماده شد اون رو برمیگردونه و اتصال بسته میشه.

Server Sent Events یا SSE

در این پروتکل وقتی کاربر درخواست میفرسته و اتصال اولیه برقرار میشه، سرور داده ها رو به صورت استریم ارسال میکنه و اتصال بسته نمیشه و اگه داده جدید هم بیاد اونا رو ارسال میکنه بدون اینکه کاربر درخواست جدیدی بفرسته در واقع به وسیله تنها یک اتصال میشه داده ها رو به صورت استریم گرفت.

اگه این لینک رو باز کنید و در مرورگر به قسمت Developer tools > Network > xhr برید و لینک stream رو انتخاب کنید میبینید که زده EventStream و میتونید ببینید که داده ها چطوری به صورت به لحظه دریافت میشه.

Web Sockets

این پروتکل مثل SSE هست و بدون اینکه اتصال بسته بشه داده ها به صورت استریم فرستاده میشه با این تفاوت که کانال ارتباطی دو طرفه هست یعنی هم کاربر میتونه داده ها رو به صورت استریم بفرسته هم سرور.

ارسال داده ها به صورت استریم در پایتون

با پروتکلهایی که گفته شد کاری نداریم الان ما میخوایم با Flask داده ها رو از سرور به سمت کاربر بفرستیم و اتصال رو نبندیم یعنی به صورت استریم

import time
from flask import Flask, Response
from werkzeug.serving import run_simple

app = Flask("SSE Server")


@app.route("/")
def index():
    def generate():
        for i in range(100):
            yield str(i)
            time.sleep(1)

    resp = Response(generate())
    return resp


run_simple(
    application=app, hostname="localhost", port=5000, use_debugger=1, use_reloader=1
)

import time

from flask import Flask, Response

from werkzeug.serving import run_simple

app = Flask("SSE Server")

@app.route("/")

def index():

def generate():

for i in range(100):

yield str(i)

time.sleep(1)

resp = Response(generate())

return resp

run_simple(

application=app, hostname="localhost", port=5000, use_debugger=1, use_reloader=1

)

الان اسکریپت بالا رو اجرا میکنیم

$ python3.6 run.py

1	$ python3.6 run.py

اگه با curl درخواست بزنید میتونید داده ها رو به صورت استریم بگیرید

$ curl http://localhost:5000

1	$ curl http://localhost:5000

ولی بلافاصله چیزی نمیگیرد

نکته ای هست اینه که وقتی خروجی رو سمت کاربر میفرستید باید انتهاش n\ یا n\n\ بزارید تا برنامه ای که باهاش داده ها رو میگیرید، که میتونه curl یا مرورگر باشه متوجه بشه که یک بسته کامل داده گرفته و نشونش بده.

برای curl باید n\n\ و برای مرورگر باید n\ بزارید که میشه

yield str(i) + '\n\n'
# or
yield str(i) + '\n'

yield str(i) + '\n\n'

# or

yield str(i) + '\n'

دقت کنید که ما هنوز وارد SSE نشدیم.

حالا موقعی رو در نظر بگیرید که داده ای برای ارسال کردن نباشه و اتصال باید همینطوری باز بمونه، به نظرتون اتصال چقدر میتونه بیکار بمونه، یعنی چیزی نفرسته و قطع هم نشه؟

هر اتصال حدودا میتونه تا 2 دقیقه بیکار بمونه و چیزی نفرسته ولی بعدش توسط فایروال هایی که در لایه های پایینی شبکه هستند قطع میشه، برای جلوگیری از قطع شدن اتصال میتونید برای مثال هر 30 ثانیه یک پیام الکی بفرستید تا اتصال زنده بمونه.

البته در شبکه های موبایل گویا همچین چیزی لازم نیست چون اتصال idle هیچ موقع بسته نمیشه و با ارسال این پیام تنها باعث مصرف بیشتر باتری و پهنای باند میشید.

خوب تا الان دیدیم چطوری داده ها رو استریم بفرستیم، حالا SSE چیه؟

ارسال استریم داده ها به صورت Server Sent Events

پروتوکل SSE یه سری استاندارد هست که میگه

سرور برای ارسال داده ها، باید هدر Content-Type: text/event-stream رو ارسال کنه

که تو Flask اینجوری میتونید اینکارو انجام بدید

resp.headers['Content-Type'] = 'text/event-stream'

1	resp.headers['Content-Type'] = 'text/event-stream'

پیام ارسالی باید حاوی رشته :data باشه که با n\n\ تموم بشه اینجوری مرورگر میفهمه که یک پیام کامل گرفته
برای دریافت پیام سمت مرورگر باید یک EventSource تعریف کرد که این پیام ها رو بخونه

به کل این میگن SSE

هر پیام میتونه فیلد های زیر رو هم داشته باشه که اختیاری هستند

id شماره پیام
event نوع پیام
retry به مرورگر میگه که اگه اتصال قطع شد بعد از گذشت اینقدر میلی ثانیه دوباره وصل بشه، بهتر هست که این مقدار تنها ابتدای اتصال یکبار فرستاده بشه و نه در هر پیام

فیلد data رو هم میتونید با n\ جدا کنید برای مثال اگه این پیام فرستاده بشه

data: I am\n
data: Pepsi\n\n

1 2	data: I am\n data: Pepsi\n\n

کاربر این رشته میگیره I am\nPepsi

چند نمونه از پیام های SSE

id: 6598\n
data: test1\n
event: sse\n
retry: 3000\n\n

data: {\n
data: "msg": "I am Pepsi"\n
data: }\n\n

id: 6598\n

data: test1\n

event: sse\n

retry: 3000\n\n

data: {\n

data: "msg": "I am Pepsi"\n

data: }\n\n

برای سمت مرورگر هم داریم

var source = new EventSource("stream_url"); 

source.onopen = function(e) { // connection was opened. }

source.onmessage = function (e) {  
  console.log(e.id, e.data);
  if (e.event == 'close') {
       source.close()
}

var source = new EventSource("stream_url");

source.onopen = function(e) { // connection was opened. }

source.onmessage = function (e) {

console.log(e.id, e.data);

if (e.event == 'close') {

source.close()

}

به جای stream_url ادرسی سروری که داده ها رو به صورت استریم میفرسته بزارید.

همه پیام ها توسط source.onmesage پردازش میشه.

پیام هایی که اولشون : داشته باشن پیام در نظر گرفته نمیشن و توسط EventSource رد میشن، این بیشتر برای مواقعی استفاده میشه که بخوایم یه سیگنال برای زنده نگه داشتن ارتباط ارسال کنیم مثل : یا keepalive: یا heartbeat:

هر موقع اتصال به هر دلیلی قطع بشه خود مرورگر تلاش میکنه تا دوباره اتصال رو برقرار کنه (در صورتی که خود سرور دستور قطع رو ارسال نکرده باشه)، اگه مقدار retry ارسال شده باشه به همون اندازه صبر میکنه و دوباره برای برقراری ارتباط تلاش میکنه، اگه نداشته باشه حدودا 3 ثانیه بعد دوباره اینکارو میکنه

وقتی اتصال از سمت مرورگر قطع بشه، مرورگر id اخرین پیام رو در هدر Last-Event-ID قرار میده و برای اتصال دوباره درخواست میزنه تا اگه سمت سرور همچین ویژگی پیاده شده باشه، ادامه پیام ها رو بگیره.

مثال سرور ارسال داده ها در Python Falcon به صورت SSE

import time

import falcon
from werkzeug.serving import run_simple

app = falcon.API()


class SSEHelper:
    @staticmethod
    def make_message(data):
        return "data: {data}\n\n".format(data=data)

    @staticmethod
    def make_heartbeat_signal():
        # all messages stating with `:` will be skipped in event source on browser side
        return ":keepalive\n\n"


class IndexResource:
    def on_get(self, req, resp):
        class Generate:
            def __init__(self):
                i = 0
                self.ids = list(range(100))

            def __iter__(self):
                return self

            def __next__(self):
                time.sleep(1)
                return SSE.make_message(str(self.ids.pop())).encode("utf-8")

        resp.stream = Generate()
        resp.connection = "keep-alive"
        resp.content_type = "text/event-stream"
        resp.cache_control = "no-cache"


app.add_route("/", IndexResource())
run_simple(
    application=app, hostname="localhost", port=5000, use_debugger=1, use_reloader=1
)

import time

import falcon

from werkzeug.serving import run_simple

app = falcon.API()

class SSEHelper:

@staticmethod

def make_message(data):

return "data: {data}\n\n".format(data=data)

@staticmethod

def make_heartbeat_signal():

# all messages stating with `:` will be skipped in event source on browser side

return ":keepalive\n\n"

class IndexResource:

def on_get(self, req, resp):

class Generate:

def __init__(self):

i = 0

self.ids = list(range(100))

def __iter__(self):

return self

def __next__(self):

time.sleep(1)

return SSE.make_message(str(self.ids.pop())).encode("utf-8")

resp.stream = Generate()

resp.connection = "keep-alive"

resp.content_type = "text/event-stream"

resp.cache_control = "no-cache"

app.add_route("/", IndexResource())

run_simple(

application=app, hostname="localhost", port=5000, use_debugger=1, use_reloader=1

)

تو Falcon برای اینکه داده ها رو به صورت استریم بفرستیم به resp.stream باید یک Iterator بدیم که مقدار بایت برگردونه

منابع

1 2 3 4