Saya memiliki aplikasi sederhana, dengan dua fungsi, satu untuk mendengarkan topik dan lainnya untuk titik akhir web. Saya ingin membuat streaming acara sisi server (SSE) yaitu teks / event-stream, sehingga pada klien saya dapat mendengarkannya menggunakan EventSource.
Saya memiliki kode berikut untuk saat ini, di mana setiap fungsi melakukan tugas khususnya:
import faust
from faust.web import Response
app = faust.App("app1", broker="kafka://localhost:29092", value_serializer="raw")
test_topic = app.topic("test")
@app.agent(test_topic)
async def test_topic_agent(stream):
async for value in stream:
print(f"test_topic_agent RECEIVED -- {value!r}")
yield value
@app.page("/")
async def index(self, request):
return self.text("yey")
Sekarang, saya ingin di indeks, seperti kode ini, tetapi menggunakan faust:
import asyncio
from aiohttp import web
from aiohttp.web import Response
from aiohttp_sse import sse_response
from datetime import datetime
async def hello(request):
loop = request.app.loop
async with sse_response(request) as resp:
while True:
data = 'Server Time : {}'.format(datetime.now())
print(data)
await resp.send(data)
await asyncio.sleep(1, loop=loop)
return resp
async def index(request):
d = """
<html>
<body>
<script>
var evtSource = new EventSource("/hello");
evtSource.onmessage = function(e) {
document.getElementById('response').innerText = e.data
}
</script>
<h1>Response from server:</h1>
<div id="response"></div>
</body>
</html>
"""
return Response(text=d, content_type='text/html')
app = web.Application()
app.router.add_route('GET', '/hello', hello)
app.router.add_route('GET', '/', index)
web.run_app(app, host='127.0.0.1', port=8080)
Saya sudah mencoba ini:
import faust
from faust.web import Response
app = faust.App("app1", broker="kafka://localhost:29092", value_serializer="raw")
test_topic = app.topic("test")
# @app.agent(test_topic)
# async def test_topic_agent(stream):
# async for value in stream:
# print(f"test_topic_agent RECEIVED -- {value!r}")
# yield value
@app.page("/", name="t1")
@app.agent(test_topic, name="t")
async def index(self, request):
return self.text("yey")
Tapi itu memberi saya kesalahan berikut:
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 299, in find_app
val = symbol_by_name(app, imp=imp)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 262, in symbol_by_name
module = imp( # type: ignore
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 376, in import_from_cwd
return imp(module, package=package)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
File "<frozen importlib._bootstrap>", line 991, in _find_and_load
File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 783, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/Users/maverick/company/demo1/baiohttp-demo/app1.py", line 18, in <module>
async def index(self, request):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/app/base.py", line 1231, in _decorator
view = view_base.from_handler(cast(ViewHandlerFun, fun))
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/web/views.py", line 50, in from_handler
return type(fun.__name__, (cls,), {
AttributeError: 'Agent' object has no attribute '__name__'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/faust_demo/bin/faust", line 8, in <module>
sys.exit(cli())
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/click/core.py", line 829, in __call__
return self.main(*args, **kwargs)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/click/core.py", line 781, in main
with self.make_context(prog_name, args, **extra) as ctx:
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 407, in make_context
self._maybe_import_app()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 372, in _maybe_import_app
find_app(appstr)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 303, in find_app
val = imp(app)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 376, in import_from_cwd
return imp(module, package=package)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
File "<frozen importlib._bootstrap>", line 991, in _find_and_load
File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 783, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/Users/maverick/company/demo1/baiohttp-demo/app1.py", line 18, in <module>
async def index(self, request):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/app/base.py", line 1231, in _decorator
view = view_base.from_handler(cast(ViewHandlerFun, fun))
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/web/views.py", line 50, in from_handler
return type(fun.__name__, (cls,), {
AttributeError: 'Agent' object has no attribute '__name__'
Saya acara mencoba ini:
import faust
from faust.web import Response
app = faust.App("app1", broker="kafka://localhost:29092", value_serializer="raw")
test_topic = app.topic("test")
# @app.agent(test_topic)
# async def test_topic_agent(stream):
# async for value in stream:
# print(f"test_topic_agent RECEIVED -- {value!r}")
# yield value
@app.agent(test_topic, name="t")
@app.page("/", name="t1")
async def index(self, request):
return self.text("yey")
Tapi saya mendapatkan kesalahan berikut:
[2020-03-28 10:32:50,676] [29976] [INFO] [^--Producer]: Creating topic 'app1-__assignor-__leader'
[2020-03-28 10:32:50,695] [29976] [INFO] [^--ReplyConsumer]: Starting...
[2020-03-28 10:32:50,695] [29976] [INFO] [^--AgentManager]: Starting...
[2020-03-28 10:32:50,695] [29976] [INFO] [^---Agent: app1.index]: Starting...
[2020-03-28 10:32:50,696] [29976] [ERROR] [^Worker]: Error: TypeError("__init__() missing 1 required positional argument: 'web'")
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/worker.py", line 273, in execute_from_commandline
self.loop.run_until_complete(self._starting_fut)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/asyncio/base_events.py", line 612, in run_until_complete
return future.result()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 767, in _actually_start
await child.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 767, in _actually_start
await child.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 760, in _actually_start
await self.on_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/manager.py", line 58, in on_start
await agent.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 760, in _actually_start
await self.on_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 282, in on_start
await self._on_start_supervisor()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 312, in _on_start_supervisor
res = await self._start_one(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 251, in _start_one
return await self._start_task(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 617, in _start_task
actor = self(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 525, in __call__
return self.actor_from_stream(stream,
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 552, in actor_from_stream
res = self.fun(actual_stream)
TypeError: __init__() missing 1 required positional argument: 'web'
[2020-03-28 10:32:50,703] [29976] [INFO] [^Worker]: Stopping...
[2020-03-28 10:32:50,703] [29976] [INFO] [^-App]: Stopping...
[2020-03-28 10:32:50,703] [29976] [INFO] [^-App]: Flush producer buffer...
[2020-03-28 10:32:50,703] [29976] [INFO] [^--TableManager]: Stopping...
Mungkinkah ada cara untuk ini? Terima kasih banyak sebelumnya!
python
python-3.x
apache-kafka
aiohttp
faust
Maverick
sumber
sumber
'Agent' object has no attribute '__name__'
... Sudahkah Anda mencoba menghapusname="t"
?name="t"
tidak membantu, kesalahan yang sama. @ cricket_007Jawaban:
Pekerja Faust juga akan mengekspos server web pada setiap contoh, yang secara default berjalan pada port 6066.
Server akan menggunakan pustaka server aiohttp HTTP dan Anda dapat memanfaatkan hal ini dan membuat streaming acara sisi-server (SSE) seperti pada kode contoh Anda.
Anda dapat membuat agen yang akan membaca dari topik Kafka
test
dan akan memperbarui variabellast_message_from_topic
dengan pesan terakhir dari topik, variabel ini akan terlihat juga dari halaman web Anda.Di halaman indeks (
@app.page('/')
) antarmuka EventSource digunakan untuk menerima acara yang dikirim oleh server. Ini terhubung ke server melalui HTTP dan menerima acara dalam format teks / event-stream dari halaman/hello
tanpa menutup koneksi.Halaman web
/hello
pada setiap detik mengirimkan teks pesan dengan pesan terakhir dari topik Kafkatest
dan dengan waktu saat ini dari server.ini
my_worker.py
kode file saya :sekarang Anda harus memulai pekerja Faust dengan perintah berikut:
di browser web Anda, Anda dapat mengakses
http://localhost:6066/
:di sini adalah kode untuk mengirim pesan ke Kafka pada topik
test
(dari file python lain):sumber
Membaca faust dokumentasi web sepertinya tidak menangani SSE.
@app.agent
dipanggil kembali ketika pesan kafka dikonsumsi dan@app.page
ketika permintaan http diproses. Menggabungkan mereka mungkin tidak mungkin.Pendekatan alternatif yang digunakan
faust.web
adalah polling dari javascript.Misalnya menggunakan:
Implementasi naif ini menyimpan pesan kafka
app.lastmsg
yang bisa didapat oleh api/msg
.Untuk menggunakan SSE, Anda dapat menggunakan
asyncio
untuk bagian web danfaust
untuk konsumen kafka.sumber