Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
mica
mica-mqtt
提交
ca44c630
mica-mqtt
项目概览
mica
/
mica-mqtt
通知
71
Star
1
Fork
1
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
mica-mqtt
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
ca44c630
编写于
12月 28, 2021
作者:
浅梦2013
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
✨
mica-mqtt client 添加批量取消订阅。
上级
89016e67
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
50 addition
and
33 deletion
+50
-33
mica-mqtt-codec/src/main/java/net/dreamlu/iot/mqtt/codec/MqttMessageBuilders.java
.../java/net/dreamlu/iot/mqtt/codec/MqttMessageBuilders.java
+21
-17
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java
...amlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java
+2
-2
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientSession.java
...reamlu/iot/mqtt/core/client/DefaultMqttClientSession.java
+2
-2
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/IMqttClientSession.java
.../net/dreamlu/iot/mqtt/core/client/IMqttClientSession.java
+2
-2
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java
...ain/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java
+17
-5
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttPendingUnSubscription.java
...eamlu/iot/mqtt/core/client/MqttPendingUnSubscription.java
+6
-5
未找到文件。
mica-mqtt-codec/src/main/java/net/dreamlu/iot/mqtt/codec/MqttMessageBuilders.java
浏览文件 @
ca44c630
...
...
@@ -21,6 +21,7 @@ import net.dreamlu.iot.mqtt.codec.MqttProperties.MqttPropertyType;
import
java.nio.ByteBuffer
;
import
java.util.ArrayList
;
import
java.util.Arrays
;
import
java.util.Collection
;
import
java.util.List
;
/**
...
...
@@ -200,22 +201,29 @@ public final class MqttMessageBuilders {
public
static
final
class
SubscribeBuilder
{
private
List
<
MqttTopicSubscription
>
subscriptions
;
private
final
List
<
MqttTopicSubscription
>
subscriptions
;
private
int
messageId
;
private
MqttProperties
properties
;
SubscribeBuilder
()
{
subscriptions
=
new
ArrayList
<>(
5
);
}
public
SubscribeBuilder
addSubscription
(
MqttQoS
qos
,
String
topic
)
{
ensureSubscriptionsExist
();
subscriptions
.
add
(
new
MqttTopicSubscription
(
topic
,
qos
));
public
SubscribeBuilder
addSubscription
(
MqttTopicSubscription
subscription
)
{
subscriptions
.
add
(
subscription
);
return
this
;
}
public
SubscribeBuilder
addSubscription
(
MqttQoS
qos
,
String
topic
)
{
return
addSubscription
(
new
MqttTopicSubscription
(
topic
,
qos
));
}
public
SubscribeBuilder
addSubscription
(
String
topic
,
MqttSubscriptionOption
option
)
{
ensureSubscriptionsExist
();
subscriptions
.
add
(
new
MqttTopicSubscription
(
topic
,
option
));
return
addSubscription
(
new
MqttTopicSubscription
(
topic
,
option
));
}
public
SubscribeBuilder
addSubscriptions
(
Collection
<
MqttTopicSubscription
>
subscriptionColl
)
{
subscriptions
.
addAll
(
subscriptionColl
);
return
this
;
}
...
...
@@ -237,31 +245,27 @@ public final class MqttMessageBuilders {
MqttSubscribePayload
mqttSubscribePayload
=
new
MqttSubscribePayload
(
subscriptions
);
return
new
MqttSubscribeMessage
(
mqttFixedHeader
,
mqttVariableHeader
,
mqttSubscribePayload
);
}
private
void
ensureSubscriptionsExist
()
{
if
(
subscriptions
==
null
)
{
subscriptions
=
new
ArrayList
<>(
5
);
}
}
}
public
static
final
class
UnsubscribeBuilder
{
private
List
<
String
>
topicFilters
;
private
final
List
<
String
>
topicFilters
;
private
int
messageId
;
private
MqttProperties
properties
;
UnsubscribeBuilder
()
{
topicFilters
=
new
ArrayList
<>(
5
);
}
public
UnsubscribeBuilder
addTopicFilter
(
String
topic
)
{
if
(
topicFilters
==
null
)
{
topicFilters
=
new
ArrayList
<>(
5
);
}
topicFilters
.
add
(
topic
);
return
this
;
}
public
UnsubscribeBuilder
addTopicFilters
(
Collection
<
String
>
topicColl
)
{
topicFilters
.
addAll
(
topicColl
);
return
this
;
}
public
UnsubscribeBuilder
messageId
(
int
messageId
)
{
this
.
messageId
=
messageId
;
return
this
;
...
...
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientProcessor.java
浏览文件 @
ca44c630
...
...
@@ -197,11 +197,11 @@ public class DefaultMqttClientProcessor implements IMqttClientProcessor {
return
;
}
if
(
logger
.
isInfoEnabled
())
{
logger
.
info
(
"MQTT Topic:{} successfully unSubscribed messageId:{}"
,
pendingUnSubscription
.
getTopic
(),
messageId
);
logger
.
info
(
"MQTT Topic:{} successfully unSubscribed messageId:{}"
,
pendingUnSubscription
.
getTopic
s
(),
messageId
);
}
pendingUnSubscription
.
onUnSubAckReceived
();
clientSession
.
removePaddingUnSubscribe
(
messageId
);
clientSession
.
removeSubscriptions
(
pendingUnSubscription
.
getTopic
());
clientSession
.
removeSubscriptions
(
pendingUnSubscription
.
getTopic
s
());
}
@Override
...
...
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/DefaultMqttClientSession.java
浏览文件 @
ca44c630
...
...
@@ -116,8 +116,8 @@ public final class DefaultMqttClientSession implements IMqttClientSession {
}
@Override
public
void
removeSubscriptions
(
String
topicFilter
)
{
subscriptions
.
remove
(
topicFilter
);
public
void
removeSubscriptions
(
List
<
String
>
topicFilters
)
{
topicFilters
.
forEach
(
subscriptions:
:
remove
);
}
@Override
...
...
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/IMqttClientSession.java
浏览文件 @
ca44c630
...
...
@@ -88,9 +88,9 @@ public interface IMqttClientSession {
/**
* 删除订阅过程消息
*
* @param topicFilter
topicFilter
* @param topicFilter
s topicFilter 集合
*/
void
removeSubscriptions
(
String
topicFilter
);
void
removeSubscriptions
(
List
<
String
>
topicFilters
);
/**
* 添加取消订阅过程消息
...
...
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java
浏览文件 @
ca44c630
...
...
@@ -25,6 +25,8 @@ import org.tio.client.TioClient;
import
org.tio.core.Tio
;
import
java.nio.ByteBuffer
;
import
java.util.Arrays
;
import
java.util.List
;
import
java.util.Objects
;
import
java.util.concurrent.ScheduledThreadPoolExecutor
;
...
...
@@ -122,18 +124,28 @@ public final class MqttClient {
/**
* 取消订阅
*
* @param topicFilter topicFilter
* @param topicFilters topicFilter 集合
* @return MqttClient
*/
public
MqttClient
unSubscribe
(
String
...
topicFilters
)
{
return
unSubscribe
(
Arrays
.
asList
(
topicFilters
));
}
/**
* 取消订阅
*
* @param topicFilters topicFilter 集合
* @return MqttClient
*/
public
MqttClient
unSubscribe
(
String
topicFilter
)
{
public
MqttClient
unSubscribe
(
List
<
String
>
topicFilters
)
{
int
messageId
=
MqttClientMessageId
.
getId
();
MqttUnsubscribeMessage
message
=
MqttMessageBuilders
.
unsubscribe
()
.
addTopicFilter
(
topicFilter
)
.
addTopicFilter
s
(
topicFilters
)
.
messageId
(
messageId
)
.
build
();
MqttPendingUnSubscription
pendingUnSubscription
=
new
MqttPendingUnSubscription
(
topicFilter
,
message
);
MqttPendingUnSubscription
pendingUnSubscription
=
new
MqttPendingUnSubscription
(
topicFilter
s
,
message
);
Boolean
result
=
Tio
.
send
(
context
,
message
);
logger
.
info
(
"MQTT Topic:{} messageId:{} unSubscribing result:{}"
,
topicFilter
,
messageId
,
result
);
logger
.
info
(
"MQTT Topic:{} messageId:{} unSubscribing result:{}"
,
topicFilter
s
,
messageId
,
result
);
// 解绑 subManage listener
clientSession
.
addPaddingUnSubscribe
(
messageId
,
pendingUnSubscription
);
pendingUnSubscription
.
startRetransmissionTimer
(
executor
,
msg
->
Tio
.
send
(
context
,
msg
));
...
...
mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttPendingUnSubscription.java
浏览文件 @
ca44c630
...
...
@@ -4,6 +4,7 @@ import net.dreamlu.iot.mqtt.codec.MqttMessage;
import
net.dreamlu.iot.mqtt.codec.MqttUnsubscribeMessage
;
import
net.dreamlu.iot.mqtt.core.common.RetryProcessor
;
import
java.util.List
;
import
java.util.concurrent.ScheduledThreadPoolExecutor
;
import
java.util.function.Consumer
;
...
...
@@ -11,16 +12,16 @@ import java.util.function.Consumer;
* MqttPendingSubscription,参考于 netty-mqtt-client
*/
final
class
MqttPendingUnSubscription
{
private
final
String
topic
;
private
final
List
<
String
>
topics
;
private
final
RetryProcessor
<
MqttUnsubscribeMessage
>
retryProcessor
=
new
RetryProcessor
<>();
MqttPendingUnSubscription
(
String
topic
,
MqttUnsubscribeMessage
unSubscribeMessage
)
{
this
.
topic
=
topic
;
MqttPendingUnSubscription
(
List
<
String
>
topics
,
MqttUnsubscribeMessage
unSubscribeMessage
)
{
this
.
topic
s
=
topics
;
this
.
retryProcessor
.
setOriginalMessage
(
unSubscribeMessage
);
}
protected
String
getTopic
()
{
return
topic
;
protected
List
<
String
>
getTopics
()
{
return
topic
s
;
}
protected
void
startRetransmissionTimer
(
ScheduledThreadPoolExecutor
executor
,
Consumer
<
MqttMessage
>
sendPacket
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录