[Feature #4793] Support MQTT protocol #4794

karsonto · 2024-03-19T09:55:19Z

Fixes #4793

Motivation

Explain the content here.
Explain why you want to make the changes and what problem you're trying to solve.

Modifications

Describe the modifications you've done.

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
If a feature is not applicable for documentation, explain why?
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

codecov · 2024-03-19T10:21:23Z

Codecov Report

Attention: Patch coverage is 0.23256% with 429 lines in your changes are missing coverage. Please review.

Project coverage is 16.11%. Comparing base (0250a89) to head (44e61e9).
Report is 7 commits behind head on master.

Files	Patch %	Lines
...ntime/core/protocol/mqtt/client/ClientManager.java	0.00%	125 Missing ⚠️
...he/eventmesh/runtime/boot/EventMeshMQTTServer.java	0.00%	86 Missing ⚠️
...re/protocol/mqtt/processor/SubscrubeProcessor.java	0.00%	48 Missing ⚠️
...me/core/protocol/mqtt/exception/MqttException.java	0.00%	31 Missing ⚠️
...rotocol/mqtt/processor/ClientConnectProcessor.java	0.00%	30 Missing ⚠️
...core/protocol/mqtt/processor/PublishProcessor.java	0.00%	28 Missing ⚠️
...eventmesh/runtime/boot/EventMeshMqttBootstrap.java	0.00%	19 Missing ⚠️
.../protocol/mqtt/processor/UnSubscrubeProcessor.java	0.00%	16 Missing ⚠️
...protocol/mqtt/processor/AbstractMqttProcessor.java	0.00%	11 Missing ⚠️
.../protocol/mqtt/processor/HealthCheckProcessor.java	0.00%	11 Missing ⚠️
... and 6 more

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #4794       +/-   ##
=============================================
+ Coverage          0   16.11%   +16.11%     
- Complexity        0     1734     +1734     
=============================================
  Files             0      870      +870     
  Lines             0    31682    +31682     
  Branches          0     2739     +2739     
=============================================
+ Hits              0     5106     +5106     
- Misses            0    26097    +26097     
- Partials          0      479      +479

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

pandaapo · 2024-03-19T10:16:57Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+
+
+    protected final transient Map<MqttMessageType, MqttProcessor> processorTable =
+        new ConcurrentHashMap<>(64);


The class is not serializable, is the transient keyword redundant?

Noted，please review again.

pandaapo · 2024-03-19T10:24:54Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+                } catch (Exception ex) {
+                    log.error("EventMeshMQTTServer RemotingServer shutdown Err!", ex);
+                }
+                System.exit(-1);


Is it appropriate to exit the process when the MQTT server fails to start?

This is a reference to other EM protocols.If you want to modify it after the discussion, I will ignore this startup failure.

@xwm1992

Currently tcp protocol, http protocol server must be started successfully, grpc protocol did not so. Now added MQTT protocol server must be started successfullyt, please community to give advice, I can not decide.

目前tcp协议、http协议的server必须成功启动，grpc协议没有如此，现在新增MQTT协议的server是否必须启动成功，请社区给出意见，我权衡不好。

我理解这里的必须启动成功实际上是受是否开启MQTT协议的配置控制的吧？这里的退出我认为没有问题，如果MQTT协议加载有问题退出了，那其实可以在配置中移除MQTT协议，保证TCP、HTTP等协议正常启动服务就好。

pandaapo · 2024-03-19T10:43:29Z

...time/src/main/java/org/apache/eventmesh/runtime/core/protocol/mqtt/client/ClientManager.java

+        this.cleanThread = new Thread(() -> {
+            while (true && !Thread.currentThread().isInterrupted()) {
+                try {
+                    ClientInfo clientInfo = delayQueue.take();


What is the role of delayQueue?

The role of delayQueue is to clean up the time out connection.

Is it up to the client to decide and set the length of time that each client's connection survives when it sends a message?

每个客户端的连接的存活时长，是靠客户端发送消息时自己决定并设置好吗？

客户端可以携带keep alive time参数，用做会话存活时间

pandaapo · 2024-03-19T10:44:35Z

...rc/main/java/org/apache/eventmesh/runtime/core/protocol/mqtt/processor/PublishProcessor.java

+
+    @Override
+    public void process(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws MqttException {
+        //


Remove this.

Noted，please review again.

pandaapo · 2024-03-21T11:19:35Z

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshMQTTServer.java

+    protected final Map<MqttMessageType, MqttProcessor> processorTable =
+        new ConcurrentHashMap<>(64);
+
+    private final transient AtomicBoolean started = new AtomicBoolean(false);


Same problem.
I'm not familiar with MQTT, so the rest of review work needs the community to complete.

Same problem.

Waiting for your response.

what's the problem here？

#4794 (comment)

Noted，please review again.

xwm1992 · 2024-03-26T02:51:12Z

@karsonto
I don’t understand the implementation of the publish processor, and I don’t see how messages of the mqtt protocol are converted and sent to eventmesh storage.

我没有理解对于publish processor的实现，没有看到将mqtt协议的消息进行转换后发送到eventmesh storage。

karsonto · 2024-03-26T02:54:38Z

@xwm1992
This PR is the framework design of the MQTT protocol , and will continue to optimize persist messages in another PR.

karsonto · 2024-03-26T07:33:48Z

This PR is the framework design of the MQTT protocol , and will continue to optimize persist messages in another PR.

Pil0tXia

Is eventmesh-protocol-plugin implementation the only work left for this protocol in Runtime?

karsonto · 2024-04-15T01:30:14Z

Is eventmesh-protocol-plugin implementation the only work left for this protocol in Runtime?

I think there is unnecessry to implement this module for this time. If it needs to be implemented in the future, I will submit a PR to supplement it.

init mqtt function

d0b3678

pandaapo reviewed Mar 19, 2024

View reviewed changes

some enhance.

1f8cf13

pandaapo reviewed Mar 21, 2024

View reviewed changes

some enhance.

44e61e9

Pil0tXia reviewed Apr 14, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature #4793] Support MQTT protocol #4794

[Feature #4793] Support MQTT protocol #4794

karsonto commented Mar 19, 2024

codecov bot commented Mar 19, 2024 •

edited

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 21, 2024

xwm1992 Mar 26, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 26, 2024

karsonto Mar 26, 2024

pandaapo Mar 19, 2024

karsonto Mar 21, 2024

pandaapo Mar 21, 2024

pandaapo Mar 26, 2024

karsonto Mar 26, 2024

pandaapo Mar 27, 2024

karsonto Mar 28, 2024

xwm1992 commented Mar 26, 2024

karsonto commented Mar 26, 2024 •

edited

karsonto commented Mar 26, 2024

Pil0tXia left a comment

karsonto commented Apr 15, 2024 •

edited



		protected final transient Map<MqttMessageType, MqttProcessor> processorTable =
		new ConcurrentHashMap<>(64);

[Feature #4793] Support MQTT protocol #4794

Are you sure you want to change the base?

[Feature #4793] Support MQTT protocol #4794

Conversation

karsonto commented Mar 19, 2024

Motivation

Modifications

Documentation

codecov bot commented Mar 19, 2024 • edited

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xwm1992 commented Mar 26, 2024

karsonto commented Mar 26, 2024 • edited

karsonto commented Mar 26, 2024

Pil0tXia left a comment

Choose a reason for hiding this comment

karsonto commented Apr 15, 2024 • edited

codecov bot commented Mar 19, 2024 •

edited

karsonto commented Mar 26, 2024 •

edited

karsonto commented Apr 15, 2024 •

edited