apache / apache/iotdb

Allow Bridging from existing MQTT Brokers to internal service

Open
#3,280 2 comments 0 reactions 0 assignees View on GitHub
Module - MQTT
Dominant language
Java
Stars
6.4k
Forks
1.2k
Avg merge
1d 23h
Merged PRs (30d)
115

Description

**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]
```

Contributor guide

Open the contributing guide

Research direction

Reproduce the Mosquitto bridge connection against the MQTT endpoint, then start with the exception path named in NettyUtils.java, MQTTMessageLogger.java, and MqttDecoder. Done means an existing broker can connect successfully and messages flow between the external broker and the integrated service in both directions.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
databases, networking
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.