mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-21 19:21:03 +02:00
187b0e6581
* Fix invalid variable name invocation * Fix error responses in websockets * Increase websocket limits to 50MB max message. Turn on Pulsar chunking by default.
56 lines
1.1 KiB
Python
56 lines
1.1 KiB
Python
|
|
# Like ServiceRequestor, but just fire-and-forget instead of request/response
|
|
|
|
import asyncio
|
|
from pulsar.schema import JsonSchema
|
|
import uuid
|
|
import logging
|
|
|
|
from . publisher import Publisher
|
|
|
|
logger = logging.getLogger("sender")
|
|
logger.setLevel(logging.INFO)
|
|
|
|
class ServiceSender:
|
|
|
|
def __init__(
|
|
self,
|
|
pulsar_host,
|
|
request_queue, request_schema,
|
|
):
|
|
|
|
self.pub = Publisher(
|
|
pulsar_host, request_queue,
|
|
schema=JsonSchema(request_schema)
|
|
)
|
|
|
|
async def start(self):
|
|
|
|
self.pub.start()
|
|
|
|
def to_request(self, request):
|
|
raise RuntimeError("Not defined")
|
|
|
|
async def process(self, request, responder=None):
|
|
|
|
try:
|
|
|
|
await asyncio.to_thread(
|
|
self.pub.send, None, self.to_request(request)
|
|
)
|
|
|
|
if responder:
|
|
await responder({}, True)
|
|
|
|
except Exception as e:
|
|
|
|
logging.error(f"Exception: {e}")
|
|
|
|
err = { "error": str(e) }
|
|
|
|
if responder:
|
|
await responder(err, True)
|
|
|
|
return err
|
|
|
|
|