Stanislav Hejny
07f6f4569c
Unit test coverage / pytest (pull_request) Failing after 1m49s
Build and Publish Docker Image / build-and-push (push) Successful in 2m29s
Unit test coverage / pytest (push) Failing after 1m45s
/ build-and-push (push) Successful in 1m52s
CI for pypl publish / publish-lib (push) Successful in 1m44s
54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
import asyncio
|
|
|
|
from amqp.config.amq_configuration import AMQConfiguration
|
|
from amqp.model.model import AMQRoute, DataMessage
|
|
from otdemo.otdemo_adapter import OTDemoAmqpAdapter
|
|
|
|
|
|
def print_document_id_static(document_id: str) -> str:
|
|
print(f"STATIC Document ID: {document_id}")
|
|
return "STATIC PRINTED!"
|
|
|
|
|
|
class Backend:
|
|
|
|
def print_document_id(self, document_id: str) -> str:
|
|
print(f"CLASS Document ID: {document_id}")
|
|
return "CLASS PRINTED!"
|
|
|
|
|
|
async def processing_loop():
|
|
while (
|
|
not otadapter.amq_service.future.done()
|
|
): # Wait until AMQ adapter initializes RabbitMQ connection
|
|
await asyncio.sleep(0.1)
|
|
# IMPORTANT - in the next line the string arg must match the first token in route-mapping property
|
|
route: AMQRoute = otadapter.amq_service.router.find_route_by_service("otdemo-2")
|
|
generator = otadapter.amq_service.data_message_generator(route) # this is a generator function
|
|
print(f"Data message generator created: {generator}")
|
|
message = await generator.asend(None) # First value must be None to start the generator
|
|
while True:
|
|
print(f"Received message: {message}")
|
|
if message is None:
|
|
break
|
|
data_message: DataMessage = message
|
|
# This is a message processing part. You may use data_message.method() or .domain() or .path()
|
|
# to access values provided by caller in the send command() call.
|
|
document_id = data_message._base64body.decode("utf-8")
|
|
print(
|
|
f"DATA_MESSAGE Attributes: {data_message.method()} {data_message.domain()} {data_message.path()}"
|
|
)
|
|
if data_message.path() == "print_document_id_static":
|
|
print_document_id_static(document_id)
|
|
elif data_message.path() == "print_document_id":
|
|
backend = Backend()
|
|
backend.print_document_id(document_id)
|
|
message = await generator.asend(True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
otadapter = OTDemoAmqpAdapter(AMQConfiguration("otdemo.backend.properties"), [])
|
|
# Run the processing loop in the AMQ service event loop
|
|
asyncio.run_coroutine_threadsafe(processing_loop(), otadapter.amq_service.loop)
|
|
otadapter.thread.join() # if the code above is not running a loop, need to join the thread to keep program running
|