add example of jetstream
This commit is contained in:
@@ -8,7 +8,7 @@ from cProfile import run
|
|||||||
from faststream import FastStream
|
from faststream import FastStream
|
||||||
from faststream.specification.asyncapi.v3_0_0.schema import servers
|
from faststream.specification.asyncapi.v3_0_0.schema import servers
|
||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
from faststream.nats import NatsBroker, JStream
|
from faststream.nats import NatsBroker, JStream, PullSub
|
||||||
from urllib3.util import url
|
from urllib3.util import url
|
||||||
|
|
||||||
NATS_SERVERS = os.getenv('NATS_URL', 'nats://localhost:4222').split(',')
|
NATS_SERVERS = os.getenv('NATS_URL', 'nats://localhost:4222').split(',')
|
||||||
@@ -19,17 +19,23 @@ app = FastStream(broker)
|
|||||||
|
|
||||||
stream = JStream(name="orders", subjects=["orders.*"])
|
stream = JStream(name="orders", subjects=["orders.*"])
|
||||||
|
|
||||||
@broker.subscriber("orders.created", stream=stream, durable="orders-worker")
|
@broker.subscriber("orders.created",
|
||||||
|
stream=stream,
|
||||||
|
#pull_sub=PullSub(),
|
||||||
|
queue="orders-queue",
|
||||||
|
#durable="orders-worker"
|
||||||
|
) #event-processor
|
||||||
async def handle_order(message: dict):
|
async def handle_order(message: dict):
|
||||||
print("received:", message)
|
print("received:", message)
|
||||||
return {"status": "ok"}
|
return {"status": "ok"}
|
||||||
|
|
||||||
@broker.publisher("orders.created", stream="orders")
|
# @broker.publisher("orders.created", stream="orders")
|
||||||
async def publish_order():
|
# async def publish_order():
|
||||||
return {"order_id": 123, "item": "book"}
|
# return {"order_id": 123, "item": "book"}
|
||||||
|
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
|
#await publish_order()
|
||||||
await app.run()
|
await app.run()
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in New Issue
Block a user