python - 使用选择器的订阅不适用于 python - stomp.py

Question

我遇到了一个问题，即使用带有消息选择器的 stomp (stomp.py) 的 Python 订阅者没有收到它应该收到的消息。有趣的是，至少在我看来，问题出在消息的发送上，而不是订阅上。

我正在使用 ActiveMQ。

这是订阅者代码:

class Listener(object):

    def __init__(self, count):
        if count <= 0:
            count = float('inf')
        self.count = count

    def on_error(self, headers, message):
        print("=" * 72)
        print('RECEIVED AN ERROR.')
        print('Message headers:')
        pp = pprint.PrettyPrinter(indent=4)
        pp.pprint(headers)
        print('Message body:')
        print(message)

    def on_message(self, headers, message):
        print("=" * 72)
        print('Message headers:')
        pp = pprint.PrettyPrinter(indent=4)
        pp.pprint(headers)
        print('Message body:')
        print(message)

def main():
    global conn

    args = parse_args()
    conn = stomp.Connection([(args.host, args.port)])
    conn.set_listener('Listener', Listener(args.count))
    conn.start()
    conn.connect(login=args.user, passcode=args.password)

    if (args.selector):
        conn.subscribe(
            destination=args.destination,
            id=1,
            ack='auto',
            headers={'selector': args.selector}
        )
    else:
        conn.subscribe(
            destination=args.destination,
            id=1,
            ack='auto'
        )

现在我可以使用诸如“type = 'test'”之类的选择器来运行这个订阅者。

如果我使用 Java JMS 发布一条消息，消息会被很好地接收。但是，如果我从 Python 发布相同的消息，则不是。

这是相关的 Python 发布代码:

headers = {}
headers['type'] = 'test'

conn = stomp.Connection12([(args.host, args.port)], auto_content_length=False)
conn.start()
conn.connect(login=args.user, passcode=args.password)
conn.send(body=body, headers=headers, destination=args.destination)
conn.disconnect()
print 'Message sent.'

我的测试和调试中的一些有趣的笔记:

1. 运行带有选择器的订阅者接收从 Java JMS 而不是从 Python 发送的匹配消息。
2. 运行不带选择器的订阅者会收到从 Java 发送的消息以及从 Python 发送的消息。

Answer 1

相当老了，但我目前正面临同样的问题，所以我想在这里留下一个可能的解决方案。

起初，according to the documentation ，您可以提供一个名为 selector 的字段，其语法类似于 SQL，并且应该是 headers 的一部分。在你的例子中:

headers = {}
headers['selector'] = "type='test'"
conn = stomp.Connection12([(args.host, args.port)], auto_content_length=False)
conn.start()
conn.connect(login=args.user, passcode=args.password)
conn.send(body=body, headers=headers, destination=args.destination)
conn.disconnect()
print 'Message sent.'

我也遇到了错误，我无法收到 JMS 发送的任何消息，但经过大量阅读后，我发现 here ，有一个字段名称 JMSType。我将代码更改为

headers['selector'] = "type='test' OR JMSType='test'"

有了 JMSType，一切都按预期进行。希望对某人有所帮助