Allow Bridging from existing MQTT Brokers to internal service
- Ngôn ngữ chính
- Java
- Star
- 6.4k
- Fork
- 1.2k
- Merge trung bình
- 1 ngày 23 giờ
- Pull request đã merge (30 ngày)
- 115
Mô tả
**Is your feature request related to a problem? Please describe.**
I like the idea of the integrated mqtt service. But I already have configured all my devices to a dedicated broker (mosquitto). I am now looking for an easy way to bridge my existing broker to the integrated mqtt service.
**Describe the solution you'd like**
Allow bridging from other brokers like e.g. Mosquitto
**Describe alternatives you've considered**
Create a dedicated client that subscribes to # and forwards all messages to IOTDB and vice versa.
**Additional context**
Mosquitto.conf:
persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log
connection iotdbltbridge
address 192.168.1.13
topic #
remote_username root
remote_password ****
When trying to setup a bridge on mosquitto side I get an exception on IOTDB side:
```
021-05-27 09:40:58,357 [nioEventLoopGroup-3-5] ERROR i.m.b.NewNettyMQTTHandler:87 - Unexpected exception while processing MQTT message. Closing Netty channel. CId=null
java.io.IOException: invalid massage
at io.moquette.broker.NettyUtils.validateMessage(NettyUtils.java:86)
at io.moquette.broker.metrics.MQTTMessageLogger.logMQTTMessage(MQTTMessageLogger.java:61)
at io.moquette.broker.metrics.MQTTMessageLogger.logMQTTMessageRead(MQTTMessageLogger.java:50)
at io.moquette.broker.metrics.MQTTMessageLogger.channelRead(MQTTMessageLogger.java:45)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.moquette.broker.metrics.MessageMetricsHandler.channelRead(MessageMetricsHandler.java:50)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:297)
at io.netty.handler.codec.ReplayingDecoder.callDecode(ReplayingDecoder.java:350)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:265)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.moquette.broker.metrics.BytesMetricsHandler.channelRead(BytesMetricsHandler.java:51)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1414)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:945)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:146)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:545)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:499)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: io.netty.handler.codec.mqtt.MqttUnacceptableProtocolVersionException: MQIsdp and -125 are not match
at io.netty.handler.codec.mqtt.MqttVersion.fromProtocolNameAndLevel(MqttVersion.java:55)
at io.netty.handler.codec.mqtt.MqttDecoder.decodeConnectionVariableHeader(MqttDecoder.java:216)
at io.netty.handler.codec.mqtt.MqttDecoder.decodeVariableHeader(MqttDecoder.java:182)
at io.netty.handler.codec.mqtt.MqttDecoder.decode(MqttDecoder.java:88)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:489)
at io.netty.handler.codec.ReplayingDecoder.callDecode(ReplayingDecoder.java:367)
... 32 common frames omitted
2021-05-27 09:41:10,498 [nioEventLoopGroup-3-4] INFO i.m.broker.MQTTConnection:253 - Notifying connection lost event. CId: mqtt-explorer-57cdbe16, channel: [id: 0x7a76b80d, L:0.0.0.0/0.0.0.0:1883]
```
Hướng dẫn đóng góp
Hướng nghiên cứu
Tái hiện kết nối bridge của Mosquitto với MQTT endpoint, sau đó bắt đầu với đường dẫn ngoại lệ được nêu trong NettyUtils.java, MQTTMessageLogger.java và MqttDecoder. Hoàn tất nghĩa là một broker hiện có có thể kết nối thành công và các message truyền theo cả hai chiều giữa broker bên ngoài và service tích hợp.
Do mô hình lập chỉ mục viết ra từ nội dung của issue.
Đánh giá
- Công nghệ
- java
- Lĩnh vực
- databases, networking
- Loại issue
- Tính năng
- Độ khó
- 5/5
- Thời gian dự kiến
- Hơn một tuần
- Mức độ hoạt động
- Đình trệ
- Độ rõ ràng
- Cần làm rõ
- Mức phù hợp với người mới
- 25/100