通过Python和azure-uamqp-python使用AMQP连接到AmazonMQ

如何解决通过Python和azure-uamqp-python使用AMQP连接到AmazonMQ

我正在尝试使用来自Azure的AmazonMQ客户端通过AMQP 1.0 protocol连接到azure-uamqp-python。

def uamqp_send_simple():
    msg_content = b"Hello world"
    parsed_uri = urlparse(URI)
    plain_auth = authentication.SASLPlain(parsed_uri.hostname,KEY_NAME,ACCESS_KEY)
    uamqp.send_message(URI,msg_content,auth=plain_auth)
    print("Message sent!")

我遇到一个Invalid Target Address错误，而uri是

ValueError: Invalid Target address: ParseResultBytes(scheme=b'amqp+ssl',netloc=xxxx.xxxx.amazonaws.com:5671',path=b'',params=b'',query=b'',fragment=b'')

使用方案amqp+ssl。

使用通过SASLPlain的身份验证方案也会发生同样的情况：

def uamqp_send_simple_sasl():

    parsed_uri = urlparse(URI)
    
    plain_auth = uamqp.authentication.SASLPlain(parsed_uri.hostname,ACCESS_KEY,port=int(PORT),transport_type=uamqp.TransportType.Amqp)
    
    target = uamqp.Target("amqp+ssl://" + str(parsed_uri.hostname))
    send_client = uamqp.SendClient(target,auth=plain_auth,debug=True,client_name="send_example_uamqp")

    try:
        
        msg_props = uamqp.message.MessageProperties()
        msg_props.to = "queue://test"
        message = uamqp.Message("content",properties=msg_props)
        
        print(message.properties)
        print(message.encode_message())

        send_client.queue_message(message)
        results = send_client.send_all_messages(close_on_done=False)
        assert not [m for m in results if m == uamqp.constants.MessageState.SendFailed]
    
    finally:
        send_client.close()
    
    print("Message sent!")