You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@qpid.apache.org by Steve Huston <sh...@riverace.com> on 2024/12/31 23:37:21 UTC
Proton python message sending
I am trying to write a python program that can receive a message, form a response, and send it back. I am getting errors on sending the response. Can someone help me see what I have wrong? The error appears to be -1 with no explanation. I'm hoping I just have some misunderstanding of how this is supposed to work.
Thank you,
-Steve
The code is:
import logging
from proton.handlers import MessagingHandler
from proton.reactor import Container
from emp import EMP_msg
from transport_echo_request import TransportEchoRequest
from transport_echo_response import TransportEchoResponse
logger = logging.getLogger("ricochet")
class Plugin(MessagingHandler):
def __init__(self):
super(Plugin, self).__init__()
def setup(self, config):
self.broker = config['broker-url']
self.recv_queue = config['recv-queue']
self.send_queue = config['send-queue']
self.message_to_send = None
def on_start(self, event):
conn = event.container.connect(self.broker)
event.container.create_receiver(conn, self.recv_queue)
self.sender = event.container.create_sender(conn, self.send_queue)
logger.info("sender %r", self.sender)
def on_message(self, event):
bytes = event.message.body # Should be transport echo request - translate?
transport = event.message.properties['TransportUsed'].tobytes().decode("utf-8")
logger.debug("body: %s", bytes.hex())
msg = EMP_msg()
ret = msg.decode(bytes)
logger.debug("EMP header decode: %s", ret)
if msg.msg_type_id != 4000 or msg.msg_ver != 1:
logger.error("Received message type %dv%d not supported.", msg.msg_type_id, msg.msg_ver)
return
request = TransportEchoRequest()
try:
request.decode(bytes)
request.decode_body()
except Exception as e:
logger.error("Decoding transport echo request: %s", e)
return
self.on_transport_echo_request(request, transport)
self.release(event.delivery)
def on_sendable(self, event):
logger.info("Sendable.")
if self.message_to_send:
event.sender.send(self.message_to_send.msg_bytes)
self.message_to_send = None
return super().on_sendable(event)
# These are local calls, not framework callbacks as the above are.
def on_transport_echo_request(self, request, transport):
logger.info("transport_echo from %s np %d on %s", request.src_addr, request.QoS_net_pref, transport)
response = TransportEchoResponse(transport, request)
response.encode_body()
response.encode()
self.sender.send(response.msg_bytes)
#self.message_to_send = response
#self.sender.offered(1)
When I run this (and the other side that initiates a request) I get:
INFO:ricochet:starting
INFO:ricochet:sender <proton._endpoints.Sender 0x10ff88c20 ~ 0x7fe6b88530a0>
INFO:proton:Connecting to Url('amqp://localhost:8000')...
INFO:proton:Connected to localhost
INFO:ricochet:Sendable.
INFO:ricochet:transport_echo from round-trip np 0 on att
Traceback (most recent call last):
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 87, in <module>
main()
~~~~^^
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 83, in main
asyncio.run(ricochet.run())
~~~~~~~~~~~^^^^^^^^^^^^^^^^
File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py", line 194, in run
return runner.run(main)
~~~~~~~~~~^^^^^^
File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py", line 721, in run_until_complete
return future.result()
~~~~~~~~~~~~~^^
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 76, in run
await asyncio.to_thread(self.run_in_thread())
~~~~~~~~~~~~~~~~~~^^
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 73, in run_in_thread
Container(self).run()
~~~~~~~~~~~~~~~~~~~^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_reactor.py", line 197, in run
while self.process():
~~~~~~~~~~~~^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_reactor.py", line 260, in process
event.dispatch(handler)
~~~~~~~~~~~~~~^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line 160, in dispatch
self.dispatch(h, type)
~~~~~~~~~~~~~^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line 160, in dispatch
self.dispatch(h, type)
~~~~~~~~~~~~~^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line 157, in dispatch
_dispatch(handler, type.method, self)
~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line 128, in _dispatch
m(*args)
~^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_handlers.py", line 256, in on_delivery
self.on_message(event)
~~~~~~~~~~~~~~~^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_handlers.py", line 281, in on_message
_dispatch(self.delegate, 'on_message', event)
~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line 128, in _dispatch
m(*args)
~^^^^^^^
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 52, in on_message
self.on_transport_echo_request(request, transport)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 68, in on_transport_echo_request
self.sender.send(response.msg_bytes)
~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py", line 1180, in send
return self.stream(obj)
~~~~~~~~~~~^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py", line 1163, in stream
return self._check(pn_link_send(self._impl, data))
~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py", line 739, in _check
raise exc("[%s]: %s" % (err, pn_error_text(pn_link_error(self._impl))))
proton._exceptions.LinkException: [-1]: None
The sender code is below. It does send the first message, but never gets a response - the above is supposed to send it but the send fails.
# Send one transport echo request and wait for a response
import logging
from proton import Message
from proton.handlers import MessagingHandler
from proton.reactor import Container
from transport_echo_request import TransportEchoRequest
from transport_echo_response import TransportEchoResponse
import emp
logger = logging.getLogger("round_trip")
class RoundTrip(MessagingHandler):
def __init__(self, server, to_ricochet_address, from_ricochet_address):
super(RoundTrip, self).__init__()
self.server = server
self.send_address = to_ricochet_address
self.recv_address = from_ricochet_address
logger.info("Setting up - broker %s", server)
def on_start(self, event):
logger.debug("on_start")
conn = event.container.connect(self.server)
event.container.create_receiver(conn, self.recv_address)
event.container.create_sender(conn, self.send_address)
def on_sendable(self, event):
logger.debug("on_sendable")
req = TransportEchoRequest()
req.data_integrity_flag = emp.EMP_INTEG_NONE
req.src_addr = "round-trip"
req.dest_addr = "ricochet"
req.guid = bytes(16)
req.encode_body()
req_bytes = req.encode()
m = Message(memoryview(req_bytes))
# The 'TransportUsed' property type was discovered by experimentation - use itcm-ping to
# send a message to the ricochet plugin and print the class of the value.
m.properties = { 'TransportUsed': memoryview(b'att') }
event.sender.send(m)
# event.sender.close()
def on_message(self, event):
logger.debug("on_message")
msg = emp.EMP_msg()
msg.decode(event.message.body)
logger.debug("msg type %d, vers %d", msg.msg_type_id, msg.msg_ver)
#assert 4001 == msg.msg_type_id
#assert 1 == msg.msg_ver
resp = TransportEchoResponse()
resp.decode(event.message.body)
resp.decode_body()
print("Response from ", resp.src_addr, " to ", resp.dest_addr, " used ", resp.sending_transport)
event.receiver.close()
event.connection.close()
logging.basicConfig(level=logging.DEBUG)
Container(RoundTrip("localhost:8000", "in", "out")).run()
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
RE: Proton python message sending
Posted by Steve Huston <sh...@riverace.com>.
Hi Ted,
Thanks very much for reading that code and responding. Very nice to hear from you.
> Have you looked at the Python Proton examples? There is a server.py in there
> that does pretty much the same thing that you are trying to do here.
> It might be helpful.
It was - I modeled my code on the examples.
> Another possible issue: You are calling sender.send with "msg_bytes" and not
> a Message object. I don't know what msg_bytes is, but I think it needs
> a send() method for it to work. Maybe you meant
> sender.send(Message(body=msg_bytes)).
<face palm>
That was it. Thank you.
Happy New Year,
-Steve
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
Re: Proton python message sending
Posted by Ted Ross <tr...@redhat.com>.
Hi Steve,
Have you looked at the Python Proton examples? There is a server.py in
there that does pretty much the same thing that you are trying to do here.
It might be helpful.
The error you are getting is a "link exception" with no description. I
suspect there's some issue with the sending queue on whatever intermediary
you are using for this task.
Typically, if you are building a server that sends responses to requests,
you will use the reply-to header in the request as the destination for the
response. You would also copy the correlation-id header from the request
to the response.
Another possible issue: You are calling sender.send with "msg_bytes" and
not a Message object. I don't know what msg_bytes is, but I think it needs
a send() method for it to work. Maybe you meant
sender.send(Message(body=msg_bytes)).
-Ted
On Tue, Dec 31, 2024 at 6:37 PM Steve Huston <sh...@riverace.com> wrote:
> I am trying to write a python program that can receive a message, form a
> response, and send it back. I am getting errors on sending the response.
> Can someone help me see what I have wrong? The error appears to be -1 with
> no explanation. I'm hoping I just have some misunderstanding of how this is
> supposed to work.
>
> Thank you,
> -Steve
>
> The code is:
>
> import logging
>
> from proton.handlers import MessagingHandler
> from proton.reactor import Container
>
> from emp import EMP_msg
> from transport_echo_request import TransportEchoRequest
> from transport_echo_response import TransportEchoResponse
>
> logger = logging.getLogger("ricochet")
>
> class Plugin(MessagingHandler):
> def __init__(self):
> super(Plugin, self).__init__()
>
> def setup(self, config):
> self.broker = config['broker-url']
> self.recv_queue = config['recv-queue']
> self.send_queue = config['send-queue']
> self.message_to_send = None
>
> def on_start(self, event):
> conn = event.container.connect(self.broker)
> event.container.create_receiver(conn, self.recv_queue)
> self.sender = event.container.create_sender(conn, self.send_queue)
> logger.info("sender %r", self.sender)
>
> def on_message(self, event):
> bytes = event.message.body # Should be transport echo request -
> translate?
> transport =
> event.message.properties['TransportUsed'].tobytes().decode("utf-8")
> logger.debug("body: %s", bytes.hex())
>
> msg = EMP_msg()
> ret = msg.decode(bytes)
> logger.debug("EMP header decode: %s", ret)
> if msg.msg_type_id != 4000 or msg.msg_ver != 1:
> logger.error("Received message type %dv%d not supported.",
> msg.msg_type_id, msg.msg_ver)
> return
> request = TransportEchoRequest()
> try:
> request.decode(bytes)
> request.decode_body()
> except Exception as e:
> logger.error("Decoding transport echo request: %s", e)
> return
> self.on_transport_echo_request(request, transport)
> self.release(event.delivery)
>
> def on_sendable(self, event):
> logger.info("Sendable.")
> if self.message_to_send:
> event.sender.send(self.message_to_send.msg_bytes)
> self.message_to_send = None
> return super().on_sendable(event)
>
> # These are local calls, not framework callbacks as the above are.
> def on_transport_echo_request(self, request, transport):
> logger.info("transport_echo from %s np %d on %s",
> request.src_addr, request.QoS_net_pref, transport)
> response = TransportEchoResponse(transport, request)
> response.encode_body()
> response.encode()
> self.sender.send(response.msg_bytes)
> #self.message_to_send = response
> #self.sender.offered(1)
>
> When I run this (and the other side that initiates a request) I get:
>
> INFO:ricochet:starting
> INFO:ricochet:sender <proton._endpoints.Sender 0x10ff88c20 ~
> 0x7fe6b88530a0>
> INFO:proton:Connecting to Url('amqp://localhost:8000')...
> INFO:proton:Connected to localhost
> INFO:ricochet:Sendable.
> INFO:ricochet:transport_echo from round-trip np 0 on att
> Traceback (most recent call last):
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 87, in
> <module>
> main()
> ~~~~^^
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 83, in
> main
> asyncio.run(ricochet.run())
> ~~~~~~~~~~~^^^^^^^^^^^^^^^^
> File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py",
> line 194, in run
> return runner.run(main)
> ~~~~~~~~~~^^^^^^
> File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py",
> line 118, in run
> return self._loop.run_until_complete(task)
> ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
> File "/usr/local/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py",
> line 721, in run_until_complete
> return future.result()
> ~~~~~~~~~~~~~^^
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 76, in
> run
> await asyncio.to_thread(self.run_in_thread())
> ~~~~~~~~~~~~~~~~~~^^
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 73, in
> run_in_thread
> Container(self).run()
> ~~~~~~~~~~~~~~~~~~~^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_reactor.py",
> line 197, in run
> while self.process():
> ~~~~~~~~~~~~^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_reactor.py",
> line 260, in process
> event.dispatch(handler)
> ~~~~~~~~~~~~~~^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line
> 160, in dispatch
> self.dispatch(h, type)
> ~~~~~~~~~~~~~^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line
> 160, in dispatch
> self.dispatch(h, type)
> ~~~~~~~~~~~~~^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line
> 157, in dispatch
> _dispatch(handler, type.method, self)
> ~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line
> 128, in _dispatch
> m(*args)
> ~^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_handlers.py",
> line 256, in on_delivery
> self.on_message(event)
> ~~~~~~~~~~~~~~~^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_handlers.py",
> line 281, in on_message
> _dispatch(self.delegate, 'on_message', event)
> ~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_events.py", line
> 128, in _dispatch
> m(*args)
> ~^^^^^^^
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 52, in
> on_message
> self.on_transport_echo_request(request, transport)
> ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
> File "/Users/user/rwn/roma/plugins/ricochet/./ricochet.py", line 68, in
> on_transport_echo_request
> self.sender.send(response.msg_bytes)
> ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py",
> line 1180, in send
> return self.stream(obj)
> ~~~~~~~~~~~^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py",
> line 1163, in stream
> return self._check(pn_link_send(self._impl, data))
> ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
> File
> "/Users/user/rwn/roma/lib/python3.13/site-packages/proton/_endpoints.py",
> line 739, in _check
> raise exc("[%s]: %s" % (err, pn_error_text(pn_link_error(self._impl))))
> proton._exceptions.LinkException: [-1]: None
>
> The sender code is below. It does send the first message, but never gets a
> response - the above is supposed to send it but the send fails.
>
> # Send one transport echo request and wait for a response
>
> import logging
> from proton import Message
> from proton.handlers import MessagingHandler
> from proton.reactor import Container
>
> from transport_echo_request import TransportEchoRequest
> from transport_echo_response import TransportEchoResponse
> import emp
>
> logger = logging.getLogger("round_trip")
>
> class RoundTrip(MessagingHandler):
> def __init__(self, server, to_ricochet_address, from_ricochet_address):
> super(RoundTrip, self).__init__()
> self.server = server
> self.send_address = to_ricochet_address
> self.recv_address = from_ricochet_address
> logger.info("Setting up - broker %s", server)
>
> def on_start(self, event):
> logger.debug("on_start")
> conn = event.container.connect(self.server)
> event.container.create_receiver(conn, self.recv_address)
> event.container.create_sender(conn, self.send_address)
>
> def on_sendable(self, event):
> logger.debug("on_sendable")
> req = TransportEchoRequest()
> req.data_integrity_flag = emp.EMP_INTEG_NONE
> req.src_addr = "round-trip"
> req.dest_addr = "ricochet"
> req.guid = bytes(16)
> req.encode_body()
> req_bytes = req.encode()
> m = Message(memoryview(req_bytes))
> # The 'TransportUsed' property type was discovered by
> experimentation - use itcm-ping to
> # send a message to the ricochet plugin and print the class of the
> value.
> m.properties = { 'TransportUsed': memoryview(b'att') }
> event.sender.send(m)
> # event.sender.close()
>
> def on_message(self, event):
> logger.debug("on_message")
> msg = emp.EMP_msg()
> msg.decode(event.message.body)
> logger.debug("msg type %d, vers %d", msg.msg_type_id, msg.msg_ver)
> #assert 4001 == msg.msg_type_id
> #assert 1 == msg.msg_ver
> resp = TransportEchoResponse()
> resp.decode(event.message.body)
> resp.decode_body()
> print("Response from ", resp.src_addr, " to ", resp.dest_addr, "
> used ", resp.sending_transport)
> event.receiver.close()
> event.connection.close()
>
> logging.basicConfig(level=logging.DEBUG)
> Container(RoundTrip("localhost:8000", "in", "out")).run()
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
> For additional commands, e-mail: users-help@qpid.apache.org
>
>