Я пытаюсь использовать клиент Paho MQTT с брокером Artemis Active MQ (образ Docker vromero / activemq-artemis), но каждый раз, когда я подписываюсь на используемую тему, я получаю сообщение «Соединение потеряно (причина: ноль)» и клиент отключается
Код клиента из официальной библиотеки Paho MQTT C с надлежащими АДРЕСОМ и ТЕМОЙ:
https://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/subasync.html
Проблема довольно странная, так как пример издателя с тем же адресом / темой работает просто отлично:
https://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/pubsync.html
После использования издателя размер связанной очереди правильный.
Настройки брокера Artemis довольно просты:
<acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true</acceptor>
...
<security-setting match="sandboxQueue">
<permission type="createNonDurableQueue" roles="userrole"/>
<permission type="deleteNonDurableQueue" roles="userrole"/>
<permission type="createDurableQueue" roles="userrole"/>
<permission type="deleteDurableQueue" roles="userrole"/>
<permission type="createAddress" roles="userrolele"/>
<permission type="deleteAddress" roles="userrole"/>
<permission type="consume" roles="userrole"/>
<permission type="browse" roles="userrole"/>
<permission type="send" roles="userrole"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="userrole"/>
</security-setting>
Мы углубились в код Артемиды и нашли следующий след:
2018-06-25 13:55:15,079 DEBUG [org.apache.activemq.artemis.core.protocol.mqtt] SESSION CREATED: 2ef3ed17-299d-4bd2-883d-063d3ae8e660
2018-06-25 13:55:15,079 INFO [org.apache.activemq.artemis.core.server.plugin.impl] AMQ841000: created connection: org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnection@e964052
2018-06-25 13:55:15,084 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843000: beforeCreateSession called with name: 9f65e1cd-786e-11e8-abe8-00155d8e1a00, username: DCUser, minLargeMessageSize: 102,400, connection: org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnection@e964052, autoCommitSends: true, autoCommitAcks: true, preAcknowledge: false, xa: false, publicAddress: null, context: OperationContextImpl [700070084] [minimalStore=9223372036854775807, storeLineUp=0, stored=0, minimalReplicated=9223372036854775807, replicationLineUp=0, replicated=0, paged=0, minimalPage=9223372036854775807, pageLineUp=0, errorCode=-1, errorMessage=null, executorsPending=0, executor=OrderedExecutor(tasks=[])]
2018-06-25 13:55:15,084 INFO [org.apache.activemq.artemis.core.server.plugin.impl] AMQ841002: created session name: 9f65e1cd-786e-11e8-abe8-00155d8e1a00, session connectionID: 596d0065
2018-06-25 13:55:15,084 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843004: beforeCreateConsumer called with ConsumerID: 106, QueueBinding: LocalQueueBinding [address=dc.config.d1, queue=QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28, filter=null, name=d1.dc.config.d1, clusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00], filterString: null, browseOnly: false, supportLargeMessage: false
2018-06-25 13:55:15,084 DEBUG [org.apache.activemq.artemis.core.server.impl.QueueImpl] QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28 adding consumer ServerConsumerImpl [id=106, filter=null, binding=LocalQueueBinding [address=dc.config.d1, queue=QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28, filter=null, name=d1.dc.config.d1, clusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]
2018-06-25 13:55:15,084 INFO [org.apache.activemq.artemis.core.server.plugin.impl] AMQ841005: created consumer with ID: 106, with session name: 9f65e1cd-786e-11e8-abe8-00155d8e1a00
2018-06-25 13:55:15,084 DEBUG [org.apache.activemq.artemis.core.server.impl.ServerSessionImpl] Session with user=DCUser, connection=org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnection@e964052 created a consumer on queue d1.dc.config.d1, filter = null
2018-06-25 13:55:15,085 DEBUG [org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl] Couldn't find any bindings for address=activemq.notifications on message=CoreMessage[messageID=108,durable=true,userID=null,priority=0, timestamp=0,expiration=0, durable=true, address=activemq.notifications,size=798,properties=TypedProperties[_AMQ_RoutingName=d1.dc.config.d1,_AMQ_Distance=0,_AMQ_ConsumerCount=1,_AMQ_User=DCUser,_AMQ_SessionName=9f65e1cd-786e-11e8-abe8-00155d8e1a00,_AMQ_Address=dc.config.d1,_AMQ_RemoteAddress=/10.254.142.17:60360,_AMQ_NotifType=CONSUMER_CREATED,_AMQ_NotifTimestamp=1529927715085,_AMQ_ClusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]@142364839
2018-06-25 13:55:15,085 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843010: beforeMessageRoute called with message: CoreMessage[messageID=108,durable=true,userID=null,priority=0, timestamp=0,expiration=0, durable=true, address=activemq.notifications,size=798,properties=TypedProperties[_AMQ_RoutingName=d1.dc.config.d1,_AMQ_Distance=0,_AMQ_ConsumerCount=1,_AMQ_User=DCUser,_AMQ_SessionName=9f65e1cd-786e-11e8-abe8-00155d8e1a00,_AMQ_Address=dc.config.d1,_AMQ_RemoteAddress=/10.254.142.17:60360,_AMQ_NotifType=CONSUMER_CREATED,_AMQ_NotifTimestamp=1529927715085,_AMQ_ClusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]@142364839, context: org.apache.activemq.artemis.core.server.impl.RoutingContextImpl@7d962ccb, direct: false, rejectDuplicates: true
2018-06-25 13:55:15,085 DEBUG [org.apache.activemq.artemis.core.postoffice.impl.PostOfficeImpl] Message CoreMessage[messageID=108,durable=true,userID=null,priority=0, timestamp=0,expiration=0, durable=true, address=activemq.notifications,size=798,properties=TypedProperties[_AMQ_RoutingName=d1.dc.config.d1,_AMQ_Distance=0,_AMQ_ConsumerCount=1,_AMQ_User=DCUser,_AMQ_SessionName=9f65e1cd-786e-11e8-abe8-00155d8e1a00,_AMQ_Address=dc.config.d1,_AMQ_RemoteAddress=/10.254.142.17:60360,_AMQ_NotifType=CONSUMER_CREATED,_AMQ_NotifTimestamp=1529927715085,_AMQ_ClusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]@142364839 is not going anywhere as it didn't have a binding on address:activemq.notifications
2018-06-25 13:55:15,085 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843011: afterMessageRoute message: CoreMessage[messageID=108,durable=true,userID=null,priority=0, timestamp=0,expiration=0, durable=true, address=activemq.notifications,size=798,properties=TypedProperties[_AMQ_RoutingName=d1.dc.config.d1,_AMQ_Distance=0,_AMQ_ConsumerCount=1,_AMQ_User=DCUser,_AMQ_SessionName=9f65e1cd-786e-11e8-abe8-00155d8e1a00,_AMQ_Address=dc.config.d1,_AMQ_RemoteAddress=/10.254.142.17:60360,_AMQ_NotifType=CONSUMER_CREATED,_AMQ_NotifTimestamp=1529927715085,_AMQ_ClusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]@142364839, with context: org.apache.activemq.artemis.core.server.impl.RoutingContextImpl@7d962ccb, direct: false, rejectDuplicates: true
2018-06-25 13:55:15,085 INFO [org.apache.activemq.artemis.core.server.plugin.impl] AMQ841010: routed message with ID: 108, result: NO_BINDINGS
2018-06-25 13:55:15,085 DEBUG [org.apache.activemq.artemis.core.server.impl.QueueImpl] QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28 doing deliver. messageReferences=1
2018-06-25 13:55:15,086 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843012: beforeDeliver called with consumer: ServerConsumerImpl [id=106, filter=null, binding=LocalQueueBinding [address=dc.config.d1, queue=QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28, filter=null, name=d1.dc.config.d1, clusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]], reference: Reference[75]:RELIABLE:CoreMessage[messageID=75,durable=true,userID=null,priority=4, timestamp=Mon Jun 25 13:46:44 CEST 2018,expiration=0, durable=true, address=dc.config.d1,size=76,properties=TypedProperties[]]@1426754715
2018-06-25 13:55:15,087 DEBUG [org.apache.activemq.artemis.core.server.plugin.impl] AMQ843013: delivered message with message ID: 75 to consumer on address: dc.config.d1, queue: d1.dc.config.d1, consumer sessionID: 9f65e1cd-786e-11e8-abe8-00155d8e1a00, consumerID: 106, full message reference: Reference[75]:RELIABLE:CoreMessage[messageID=75,durable=true,userID=null,priority=4, timestamp=Mon Jun 25 13:46:44 CEST 2018,expiration=0, durable=true, address=dc.config.d1,size=76,properties=TypedProperties[]]@1426754715, full consumer: ServerConsumerImpl [id=106, filter=null, binding=LocalQueueBinding [address=dc.config.d1, queue=QueueImpl[name=d1.dc.config.d1, postOffice=PostOfficeImpl [server=ActiveMQServerImpl::serverUUID=f8606cbe-786c-11e8-abe8-00155d8e1a00], temp=false]@4afaac28, filter=null, name=d1.dc.config.d1, clusterName=d1.dc.config.d1f8606cbe-786c-11e8-abe8-00155d8e1a00]]
2018-06-25 13:55:15,087 INFO [org.apache.activemq.artemis.core.server.plugin.impl] AMQ841012: delivered message with message ID: 75, to consumer on address: dc.config.d1, queue: d1.dc.config.d1, consumer sessionID: 9f65e1cd-786e-11e8-abe8-00155d8e1a00, consumerID: 106
2018-06-25 13:55:15,094 DEBUG [org.apache.activemq.artemis.core.protocol.mqtt] Error processing Control Packet, Disconnecting Client: ActiveMQAddressDoesNotExistException[errorType=ADDRESS_DOES_NOT_EXIST message=AMQ119203: Address Does Not Exist: $sys.mqtt.queue.qos2.d1]
at org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.createQueue(ActiveMQServerImpl.java:2890) [artemis-server-2.6.2.jar:2.6.2]
at org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.createQueue(ActiveMQServerImpl.java:1699) [artemis-server-2.6.2.jar:2.6.2]
at org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.createQueue(ActiveMQServerImpl.java:1679) [artemis-server-2.6.2.jar:2.6.2]
at org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.createQueue(ActiveMQServerImpl.java:1663) [artemis-server-2.6.2.jar:2.6.2]
at org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl.createQueue(ActiveMQServerImpl.java:1638) [artemis-server-2.6.2.jar:2.6.2]
at org.apache.activemq.artemis.core.protocol.mqtt.MQTTPublishManager.createManagementQueue(MQTTPublishManager.java:104) [artemis-mqtt-protocol-2.6.2.jar:]
at org.apache.activemq.artemis.core.protocol.mqtt.MQTTPublishManager.getManagementAddress(MQTTPublishManager.java:198) [artemis-mqtt-protocol-2.6.2.jar:]
at org.apache.activemq.artemis.core.protocol.mqtt.MQTTPublishManager.handlePubRec(MQTTPublishManager.java:208) [artemis-mqtt-protocol-2.6.2.jar:]
at org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolHandler.handlePubrec(MQTTProtocolHandler.java:208) [artemis-mqtt-protocol-2.6.2.jar:]
at org.apache.activemq.artemis.core.protocol.mqtt.MQTTProtocolHandler.channelRead(MQTTProtocolHandler.java:116) [artemis-mqtt-protocol-2.6.2.jar:]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884) [netty-all-4.1.24.Final.jar:4.1.24.Final]
at org.apache.activemq.artemis.utils.ActiveMQThreadFactory$1.run(ActiveMQThreadFactory.java:118) [artemis-commons-2.6.2.jar:2.6.2]
Есть какие-нибудь идеи, как поступить?
Спасибо!