提交 5933f060 编写于 作者: 如梦技术's avatar 如梦技术 🐛

🎉 Initial commit.

上级
# http://editorconfig.org
root = true
# 空格替代Tab缩进在各种编辑工具下效果一致
[*]
indent_style = space
indent_size = 4
charset = utf-8
end_of_line = lf
trim_trailing_whitespace = true
insert_final_newline = true
[*.java]
indent_style = tab
[*.{json,yml}]
indent_size = 2
[*.md]
insert_final_newline = false
trim_trailing_whitespace = false
### gradle ###
.gradle
!gradle/wrapper/gradle-wrapper.jar
### STS ###
.settings/
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
bin/
### IntelliJ IDEA ###
.idea
out/
*.iws
*.iml
*.ipr
### NetBeans ###
nbproject/private/
build/
nbbuild/
dist/
nbdist/
.nb-gradle/
### maven ###
target/
*.war
*.ear
*.zip
*.tar
*.tar.gz
# logs #
logs
# temp ignore
*.log
*.cache
*.diff
*.patch
*.tmp
*.java~
*.properties~
*.xml~
# system ignore
.DS_Store
Thumbs.db
Servers
.metadata
upload
gen_code
/*
* Copyright 2007-present the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import java.net.*;
import java.io.*;
import java.nio.channels.*;
import java.util.Properties;
public class MavenWrapperDownloader {
private static final String WRAPPER_VERSION = "0.5.5";
/**
* Default URL to download the maven-wrapper.jar from, if no 'downloadUrl' is provided.
*/
private static final String DEFAULT_DOWNLOAD_URL = "https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/"
+ WRAPPER_VERSION + "/maven-wrapper-" + WRAPPER_VERSION + ".jar";
/**
* Path to the maven-wrapper.properties file, which might contain a downloadUrl property to
* use instead of the default one.
*/
private static final String MAVEN_WRAPPER_PROPERTIES_PATH =
".mvn/wrapper/maven-wrapper.properties";
/**
* Path where the maven-wrapper.jar will be saved to.
*/
private static final String MAVEN_WRAPPER_JAR_PATH =
".mvn/wrapper/maven-wrapper.jar";
/**
* Name of the property which should be used to override the default download url for the wrapper.
*/
private static final String PROPERTY_NAME_WRAPPER_URL = "wrapperUrl";
public static void main(String args[]) {
System.out.println("- Downloader started");
File baseDirectory = new File(args[0]);
System.out.println("- Using base directory: " + baseDirectory.getAbsolutePath());
// If the maven-wrapper.properties exists, read it and check if it contains a custom
// wrapperUrl parameter.
File mavenWrapperPropertyFile = new File(baseDirectory, MAVEN_WRAPPER_PROPERTIES_PATH);
String url = DEFAULT_DOWNLOAD_URL;
if(mavenWrapperPropertyFile.exists()) {
FileInputStream mavenWrapperPropertyFileInputStream = null;
try {
mavenWrapperPropertyFileInputStream = new FileInputStream(mavenWrapperPropertyFile);
Properties mavenWrapperProperties = new Properties();
mavenWrapperProperties.load(mavenWrapperPropertyFileInputStream);
url = mavenWrapperProperties.getProperty(PROPERTY_NAME_WRAPPER_URL, url);
} catch (IOException e) {
System.out.println("- ERROR loading '" + MAVEN_WRAPPER_PROPERTIES_PATH + "'");
} finally {
try {
if(mavenWrapperPropertyFileInputStream != null) {
mavenWrapperPropertyFileInputStream.close();
}
} catch (IOException e) {
// Ignore ...
}
}
}
System.out.println("- Downloading from: " + url);
File outputFile = new File(baseDirectory.getAbsolutePath(), MAVEN_WRAPPER_JAR_PATH);
if(!outputFile.getParentFile().exists()) {
if(!outputFile.getParentFile().mkdirs()) {
System.out.println(
"- ERROR creating output directory '" + outputFile.getParentFile().getAbsolutePath() + "'");
}
}
System.out.println("- Downloading to: " + outputFile.getAbsolutePath());
try {
downloadFileFromURL(url, outputFile);
System.out.println("Done");
System.exit(0);
} catch (Throwable e) {
System.out.println("- Error downloading");
e.printStackTrace();
System.exit(1);
}
}
private static void downloadFileFromURL(String urlString, File destination) throws Exception {
if (System.getenv("MVNW_USERNAME") != null && System.getenv("MVNW_PASSWORD") != null) {
String username = System.getenv("MVNW_USERNAME");
char[] password = System.getenv("MVNW_PASSWORD").toCharArray();
Authenticator.setDefault(new Authenticator() {
@Override
protected PasswordAuthentication getPasswordAuthentication() {
return new PasswordAuthentication(username, password);
}
});
}
URL website = new URL(urlString);
ReadableByteChannel rbc;
rbc = Channels.newChannel(website.openStream());
FileOutputStream fos = new FileOutputStream(destination);
fos.getChannel().transferFrom(rbc, 0, Long.MAX_VALUE);
fos.close();
rbc.close();
}
}
distributionUrl=http://nexus.vava/repository/maven-public//org/apache/maven/apache-maven/3.3.3/apache-maven-3.3.3-bin.zip
wrapperUrl=http://nexus.vava/repository/maven-public//io/takari/maven-wrapper/0.5.5/maven-wrapper-0.5.5.jar
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction, and
distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by the copyright
owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all other entities
that control, are controlled by, or are under common control with that entity.
For the purposes of this definition, "control" means (i) the power, direct or
indirect, to cause the direction or management of such entity, whether by
contract or otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity exercising
permissions granted by this License.
"Source" form shall mean the preferred form for making modifications, including
but not limited to software source code, documentation source, and configuration
files.
"Object" form shall mean any form resulting from mechanical transformation or
translation of a Source form, including but not limited to compiled object code,
generated documentation, and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or Object form, made
available under the License, as indicated by a copyright notice that is included
in or attached to the work (an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object form, that
is based on (or derived from) the Work and for which the editorial revisions,
annotations, elaborations, or other modifications represent, as a whole, an
original work of authorship. For the purposes of this License, Derivative Works
shall not include works that remain separable from, or merely link (or bind by
name) to the interfaces of, the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including the original version
of the Work and any modifications or additions to that Work or Derivative Works
thereof, that is intentionally submitted to Licensor for inclusion in the Work
by the copyright owner or by an individual or Legal Entity authorized to submit
on behalf of the copyright owner. For the purposes of this definition,
"submitted" means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems, and
issue tracking systems that are managed by, or on behalf of, the Licensor for
the purpose of discussing and improving the Work, but excluding communication
that is conspicuously marked or otherwise designated in writing by the copyright
owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity on behalf
of whom a Contribution has been received by Licensor and subsequently
incorporated within the Work.
2. Grant of Copyright License.
Subject to the terms and conditions of this License, each Contributor hereby
grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
irrevocable copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the Work and such
Derivative Works in Source or Object form.
3. Grant of Patent License.
Subject to the terms and conditions of this License, each Contributor hereby
grants to You a perpetual, worldwide, non-exclusive, no-charge, royalty-free,
irrevocable (except as stated in this section) patent license to make, have
made, use, offer to sell, sell, import, and otherwise transfer the Work, where
such license applies only to those patent claims licensable by such Contributor
that are necessarily infringed by their Contribution(s) alone or by combination
of their Contribution(s) with the Work to which such Contribution(s) was
submitted. If You institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work or a
Contribution incorporated within the Work constitutes direct or contributory
patent infringement, then any patent licenses granted to You under this License
for that Work shall terminate as of the date such litigation is filed.
4. Redistribution.
You may reproduce and distribute copies of the Work or Derivative Works thereof
in any medium, with or without modifications, and in Source or Object form,
provided that You meet the following conditions:
You must give any other recipients of the Work or Derivative Works a copy of
this License; and
You must cause any modified files to carry prominent notices stating that You
changed the files; and
You must retain, in the Source form of any Derivative Works that You distribute,
all copyright, patent, trademark, and attribution notices from the Source form
of the Work, excluding those notices that do not pertain to any part of the
Derivative Works; and
If the Work includes a "NOTICE" text file as part of its distribution, then any
Derivative Works that You distribute must include a readable copy of the
attribution notices contained within such NOTICE file, excluding those notices
that do not pertain to any part of the Derivative Works, in at least one of the
following places: within a NOTICE text file distributed as part of the
Derivative Works; within the Source form or documentation, if provided along
with the Derivative Works; or, within a display generated by the Derivative
Works, if and wherever such third-party notices normally appear. The contents of
the NOTICE file are for informational purposes only and do not modify the
License. You may add Your own attribution notices within Derivative Works that
You distribute, alongside or as an addendum to the NOTICE text from the Work,
provided that such additional attribution notices cannot be construed as
modifying the License.
You may add Your own copyright statement to Your modifications and may provide
additional or different license terms and conditions for use, reproduction, or
distribution of Your modifications, or for any such Derivative Works as a whole,
provided Your use, reproduction, and distribution of the Work otherwise complies
with the conditions stated in this License.
5. Submission of Contributions.
Unless You explicitly state otherwise, any Contribution intentionally submitted
for inclusion in the Work by You to the Licensor shall be under the terms and
conditions of this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify the terms of
any separate license agreement you may have executed with Licensor regarding
such Contributions.
6. Trademarks.
This License does not grant permission to use the trade names, trademarks,
service marks, or product names of the Licensor, except as required for
reasonable and customary use in describing the origin of the Work and
reproducing the content of the NOTICE file.
7. Disclaimer of Warranty.
Unless required by applicable law or agreed to in writing, Licensor provides the
Work (and each Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied,
including, without limitation, any warranties or conditions of TITLE,
NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A PARTICULAR PURPOSE. You are
solely responsible for determining the appropriateness of using or
redistributing the Work and assume any risks associated with Your exercise of
permissions under this License.
8. Limitation of Liability.
In no event and under no legal theory, whether in tort (including negligence),
contract, or otherwise, unless required by applicable law (such as deliberate
and grossly negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special, incidental,
or consequential damages of any character arising as a result of this License or
out of the use or inability to use the Work (including but not limited to
damages for loss of goodwill, work stoppage, computer failure or malfunction, or
any and all other commercial damages or losses), even if such Contributor has
been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability.
While redistributing the Work or Derivative Works thereof, You may choose to
offer, and charge a fee for, acceptance of support, warranty, indemnity, or
other liability obligations and/or rights consistent with this License. However,
in accepting such obligations, You may act only on Your own behalf and on Your
sole responsibility, not on behalf of any other Contributor, and only if You
agree to indemnify, defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason of your
accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work
To apply the Apache License to your work, attach the following boilerplate
notice, with the fields enclosed by brackets "{}" replaced with your own
identifying information. (Don't include the brackets!) The text should be
enclosed in the appropriate comment syntax for the file format. We also
recommend that a file or class name and description of purpose be included on
the same "printed page" as the copyright notice for easier identification within
third-party archives.
Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
# mica mqtt 组件
基于 `t-io` 实现的 `mqtt` iot组件。
## 文档
- [mqtt 协议文档](https://github.com/mcxiaoke/mqtt)
## 代表
- 继续抽象,方便使用。
- 实现 `mqtt-broker` 功能。
## 参考
- [iot-mqtt-server](https://gitee.com/recallcode/iot-mqtt-server)
## 微信公众号
![如梦技术](docs/img/dreamlu-weixin.jpg)
精彩内容每日推荐!
\ No newline at end of file
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>mica-mqtt-codec</artifactId>
<dependencies>
<dependency>
<groupId>org.t-io</groupId>
<artifactId>tio-core</artifactId>
</dependency>
</dependencies>
</project>
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
/**
* ByteBuffer 工具
*
* @author L.cm
*/
public class ByteBufferUtil {
/**
* 空 byte 数组
*/
public static final byte[] EMPTY = new byte[0];
/**
* read byte
* @param buffer ByteBuffer
* @return byte
*/
public static byte readByte(ByteBuffer buffer) {
return buffer.get();
}
/**
* read unsigned byte
* @param buffer ByteBuffer
* @return short
*/
public static short readUnsignedByte(ByteBuffer buffer) {
return (short) (readByte(buffer) & 0xFF);
}
/**
* skip bytes
* @param buffer ByteBuffer
* @param skip skip bytes
* @return ByteBuffer
*/
public static ByteBuffer skipBytes(ByteBuffer buffer, int skip) {
buffer.position(buffer.position() + skip);
return buffer;
}
public static String toString(ByteBuffer buffer) {
return toString(buffer, StandardCharsets.UTF_8);
}
public static String toString(ByteBuffer buffer, Charset charset) {
return new String(buffer.array(), buffer.position(), buffer.limit(), charset);
}
public static ByteBuffer clone(ByteBuffer original) {
ByteBuffer clone = ByteBuffer.allocate(original.capacity());
// copy from the beginning
original.rewind();
clone.put(original);
original.rewind();
clone.flip();
return clone;
}
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* 解码异常
*
* @author L.cm
*/
public class DecoderException extends RuntimeException {
private static final long serialVersionUID = 1L;
public DecoderException() {
}
public DecoderException(String message) {
super(message);
}
public DecoderException(String message, Throwable cause) {
super(message, cause);
}
public DecoderException(Throwable cause) {
super(cause);
}
public DecoderException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
super(message, cause, enableSuppression, writableStackTrace);
}
}
/*
* Copyright 2012 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.Objects;
/**
* 解析结果集
*
* @author L.cm
*/
public class DecoderResult {
public static final DecoderResult SUCCESS = new DecoderResult();
private final boolean success;
private final Throwable cause;
public DecoderResult() {
this(true, null);
}
public DecoderResult(Throwable cause) {
this(false, Objects.requireNonNull(cause, "cause is null"));
}
public DecoderResult(boolean success, Throwable cause) {
this.success = success;
this.cause = cause;
}
public static DecoderResult failure(Throwable cause) {
return new DecoderResult(cause);
}
public boolean isSuccess() {
return success;
}
public boolean isFailure() {
return !success;
}
public Throwable getCause() {
return cause;
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
final class MqttCodecUtil {
private static final char[] TOPIC_WILDCARDS = {'#', '+'};
private static final int MIN_CLIENT_ID_LENGTH = 1;
private static final int MAX_CLIENT_ID_LENGTH = 23;
static boolean isValidPublishTopicName(String topicName) {
// publish topic name must not contain any wildcard
for (char c : TOPIC_WILDCARDS) {
if (topicName.indexOf(c) >= 0) {
return false;
}
}
return true;
}
static boolean isValidMessageId(int messageId) {
return messageId != 0;
}
static boolean isValidClientId(MqttVersion mqttVersion, String clientId) {
if (mqttVersion == MqttVersion.MQTT_3_1) {
return clientId != null && clientId.length() >= MIN_CLIENT_ID_LENGTH &&
clientId.length() <= MAX_CLIENT_ID_LENGTH;
}
if (mqttVersion == MqttVersion.MQTT_3_1_1) {
// In 3.1.3.1 Client Identifier of MQTT 3.1.1 specification, The Server MAY allow ClientId’s
// that contain more than 23 encoded bytes. And, The Server MAY allow zero-length ClientId.
return clientId != null;
}
throw new IllegalArgumentException(mqttVersion + " is unknown mqtt version");
}
static MqttFixedHeader validateFixedHeader(MqttFixedHeader mqttFixedHeader) {
switch (mqttFixedHeader.messageType()) {
case PUBREL:
case SUBSCRIBE:
case UNSUBSCRIBE:
if (mqttFixedHeader.qosLevel() != MqttQoS.AT_LEAST_ONCE) {
throw new DecoderException(mqttFixedHeader.messageType().name() + " message must have QoS 1");
}
default:
return mqttFixedHeader;
}
}
static MqttFixedHeader resetUnusedFields(MqttFixedHeader mqttFixedHeader) {
switch (mqttFixedHeader.messageType()) {
case CONNECT:
case CONNACK:
case PUBACK:
case PUBREC:
case PUBCOMP:
case SUBACK:
case UNSUBACK:
case PINGREQ:
case PINGRESP:
case DISCONNECT:
if (mqttFixedHeader.isDup() ||
mqttFixedHeader.qosLevel() != MqttQoS.AT_MOST_ONCE ||
mqttFixedHeader.isRetain()) {
return new MqttFixedHeader(
mqttFixedHeader.messageType(),
false,
MqttQoS.AT_MOST_ONCE,
false,
mqttFixedHeader.remainingLength());
}
return mqttFixedHeader;
case PUBREL:
case SUBSCRIBE:
case UNSUBSCRIBE:
if (mqttFixedHeader.isRetain()) {
return new MqttFixedHeader(
mqttFixedHeader.messageType(),
mqttFixedHeader.isDup(),
mqttFixedHeader.qosLevel(),
false,
mqttFixedHeader.remainingLength());
}
return mqttFixedHeader;
default:
return mqttFixedHeader;
}
}
private MqttCodecUtil() {
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connack">MQTTV3.1/connack</a>
*/
public final class MqttConnAckMessage extends MqttMessage {
public MqttConnAckMessage(MqttFixedHeader mqttFixedHeader,
MqttConnAckVariableHeader variableHeader) {
super(mqttFixedHeader, variableHeader);
}
@Override
public MqttConnAckVariableHeader variableHeader() {
return (MqttConnAckVariableHeader) super.variableHeader();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* Variable header of {@link MqttConnectMessage}
*/
public final class MqttConnAckVariableHeader {
private final MqttConnectReturnCode connectReturnCode;
private final boolean sessionPresent;
public MqttConnAckVariableHeader(MqttConnectReturnCode connectReturnCode,
boolean sessionPresent) {
this.connectReturnCode = connectReturnCode;
this.sessionPresent = sessionPresent;
}
public MqttConnectReturnCode connectReturnCode() {
return connectReturnCode;
}
public boolean isSessionPresent() {
return sessionPresent;
}
@Override
public String toString() {
return "MqttConnAckVariableHeader{" +
"connectReturnCode=" + connectReturnCode +
", sessionPresent=" + sessionPresent +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#connect">MQTTV3.1/connect</a>
*/
public final class MqttConnectMessage extends MqttMessage {
public MqttConnectMessage(MqttFixedHeader mqttFixedHeader,
MqttConnectVariableHeader variableHeader,
MqttConnectPayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttConnectVariableHeader variableHeader() {
return (MqttConnectVariableHeader) super.variableHeader();
}
@Override
public MqttConnectPayload payload() {
return (MqttConnectPayload) super.payload();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
/**
* Payload of {@link MqttConnectMessage}
*/
public final class MqttConnectPayload {
private final String clientIdentifier;
private final String willTopic;
private final byte[] willMessage;
private final String userName;
private final byte[] password;
public MqttConnectPayload(String clientIdentifier,
String willTopic,
byte[] willMessage,
String userName,
byte[] password) {
this.clientIdentifier = clientIdentifier;
this.willTopic = willTopic;
this.willMessage = willMessage;
this.userName = userName;
this.password = password;
}
public String clientIdentifier() {
return clientIdentifier;
}
public String willTopic() {
return willTopic;
}
public byte[] willMessageInBytes() {
return willMessage;
}
public String userName() {
return userName;
}
public byte[] passwordInBytes() {
return password;
}
public String password() {
return password == null ? null : new String(password, StandardCharsets.UTF_8);
}
@Override
public String toString() {
return "MqttConnectPayload{" +
"clientIdentifier='" + clientIdentifier + '\'' +
", willTopic='" + willTopic + '\'' +
", willMessage=" + Arrays.toString(willMessage) +
", userName='" + userName + '\'' +
", password=" + password() +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Return Code of {@link MqttConnAckMessage}
*/
public enum MqttConnectReturnCode {
/**
* 0 接受连接
*/
CONNECTION_ACCEPTED((byte) 0x00),
/**
* 1 连接被拒绝,协议版本不可接受
*/
CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION((byte) 0X01),
/**
* 2 连接被拒绝,标识符被拒绝
*/
CONNECTION_REFUSED_IDENTIFIER_REJECTED((byte) 0x02),
/**
* 3 连接被拒绝,服务器不可用
*/
CONNECTION_REFUSED_SERVER_UNAVAILABLE((byte) 0x03),
/**
* 4 连接被拒绝,用户名或密码错误
*/
CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD((byte) 0x04),
/**
* 5 连接被拒绝,未经授权
*/
CONNECTION_REFUSED_NOT_AUTHORIZED((byte) 0x05);
private static final Map<Byte, MqttConnectReturnCode> VALUE_TO_CODE_MAP;
static {
final Map<Byte, MqttConnectReturnCode> valueMap = new HashMap<>();
for (MqttConnectReturnCode code : values()) {
valueMap.put(code.byteValue, code);
}
VALUE_TO_CODE_MAP = Collections.unmodifiableMap(valueMap);
}
private final byte byteValue;
MqttConnectReturnCode(byte byteValue) {
this.byteValue = byteValue;
}
public byte byteValue() {
return byteValue;
}
public static MqttConnectReturnCode valueOf(byte b) {
if (VALUE_TO_CODE_MAP.containsKey(b)) {
return VALUE_TO_CODE_MAP.get(b);
}
throw new IllegalArgumentException("unknown connect return code: " + (b & 0xFF));
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* Variable Header for the {@link MqttConnectMessage}
*/
public final class MqttConnectVariableHeader {
private final String name;
private final int version;
private final boolean hasUserName;
private final boolean hasPassword;
private final boolean isWillRetain;
private final int willQos;
private final boolean isWillFlag;
private final boolean isCleanSession;
private final int keepAliveTimeSeconds;
public MqttConnectVariableHeader(
String name,
int version,
boolean hasUserName,
boolean hasPassword,
boolean isWillRetain,
int willQos,
boolean isWillFlag,
boolean isCleanSession,
int keepAliveTimeSeconds) {
this.name = name;
this.version = version;
this.hasUserName = hasUserName;
this.hasPassword = hasPassword;
this.isWillRetain = isWillRetain;
this.willQos = willQos;
this.isWillFlag = isWillFlag;
this.isCleanSession = isCleanSession;
this.keepAliveTimeSeconds = keepAliveTimeSeconds;
}
public String name() {
return name;
}
public int version() {
return version;
}
public boolean hasUserName() {
return hasUserName;
}
public boolean hasPassword() {
return hasPassword;
}
public boolean isWillRetain() {
return isWillRetain;
}
public int willQos() {
return willQos;
}
public boolean isWillFlag() {
return isWillFlag;
}
public boolean isCleanSession() {
return isCleanSession;
}
public int keepAliveTimeSeconds() {
return keepAliveTimeSeconds;
}
@Override
public String toString() {
return "MqttConnectVariableHeader{" +
"name='" + name + '\'' +
", version=" + version +
", hasUserName=" + hasUserName +
", hasPassword=" + hasPassword +
", isWillRetain=" + isWillRetain +
", willQos=" + willQos +
", isWillFlag=" + isWillFlag +
", isCleanSession=" + isCleanSession +
", keepAliveTimeSeconds=" + keepAliveTimeSeconds +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
/**
* Decodes Mqtt messages from bytes, following
* <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">
* the MQTT protocol specification v3.1</a>
*/
public final class MqttDecoder {
public static final MqttDecoder INSTANCE = new MqttDecoder();
private static final int DEFAULT_MAX_BYTES_IN_MESSAGE = 8092;
private static final int MQTT_PROTOCOL_LENGTH = 2;
private final int maxBytesInMessage;
public MqttDecoder() {
this(DEFAULT_MAX_BYTES_IN_MESSAGE);
}
public MqttDecoder(int maxBytesInMessage) {
this.maxBytesInMessage = maxBytesInMessage;
}
public boolean isSupport(ByteBuffer readBuffer) {
return readBuffer.limit() >= MQTT_PROTOCOL_LENGTH;
}
public MqttMessage decode(ByteBuffer readBuffer) {
// 1. 首先判断缓存中协议头是否读完(MQTT协议头为2字节)
if (readBuffer.limit() < MQTT_PROTOCOL_LENGTH) {
return null;
}
// 2. 解析 FixedHeader
MqttFixedHeader mqttFixedHeader;
int bytesRemainingInVariablePart;
try {
mqttFixedHeader = decodeFixedHeader(readBuffer);
bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
} catch (Exception cause) {
return MqttMessageFactory.newInvalidMessage(cause);
}
// 3. 解析头信息
Object variableHeader = null;
try {
Result<?> decodedVariableHeader = decodeVariableHeader(readBuffer, mqttFixedHeader);
variableHeader = decodedVariableHeader.value;
if (bytesRemainingInVariablePart > maxBytesInMessage) {
throw new DecoderException("too large message: " + bytesRemainingInVariablePart + " bytes");
}
bytesRemainingInVariablePart -= decodedVariableHeader.numberOfBytesConsumed;
} catch (Exception cause) {
return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
}
// 4. 解析消息体
final Result<?> decodedPayload;
try {
decodedPayload = decodePayload(readBuffer,
mqttFixedHeader.messageType(),
bytesRemainingInVariablePart,
variableHeader);
bytesRemainingInVariablePart -= decodedPayload.numberOfBytesConsumed;
if (bytesRemainingInVariablePart != 0) {
throw new DecoderException("non-zero remaining payload bytes: " +
bytesRemainingInVariablePart + " (" + mqttFixedHeader.messageType() + ')');
}
return MqttMessageFactory.newMessage(mqttFixedHeader, variableHeader, decodedPayload.value);
} catch (Throwable cause) {
return MqttMessageFactory.newInvalidMessage(mqttFixedHeader, variableHeader, cause);
}
}
/**
* Decodes the fixed header. It's one byte for the flags and then variable bytes for the remaining length.
*
* @param buffer the buffer to decode from
* @return the fixed header
*/
private static MqttFixedHeader decodeFixedHeader(ByteBuffer buffer) {
short b1 = ByteBufferUtil.readUnsignedByte(buffer);
MqttMessageType messageType = MqttMessageType.valueOf(b1 >> 4);
boolean dupFlag = (b1 & 0x08) == 0x08;
int qosLevel = (b1 & 0x06) >> 1;
boolean retain = (b1 & 0x01) != 0;
int remainingLength = 0;
int multiplier = 1;
short digit;
int loops = 0;
do {
digit = buffer.get();
remainingLength += (digit & 127) * multiplier;
multiplier *= 128;
loops++;
} while ((digit & 128) != 0 && loops < 4);
// MQTT protocol limits Remaining Length to 4 bytes
if (loops == 4 && (digit & 128) != 0) {
throw new DecoderException("remaining length exceeds 4 digits (" + messageType + ')');
}
MqttFixedHeader decodedFixedHeader = new MqttFixedHeader(messageType, dupFlag, MqttQoS.valueOf(qosLevel), retain, remainingLength);
return MqttCodecUtil.validateFixedHeader(MqttCodecUtil.resetUnusedFields(decodedFixedHeader));
}
/**
* Decodes the variable header (if any)
*
* @param buffer the buffer to decode from
* @param mqttFixedHeader MqttFixedHeader of the same message
* @return the variable header
*/
private static Result<?> decodeVariableHeader(ByteBuffer buffer, MqttFixedHeader mqttFixedHeader) {
switch (mqttFixedHeader.messageType()) {
case CONNECT:
return decodeConnectionVariableHeader(buffer);
case CONNACK:
return decodeConnAckVariableHeader(buffer);
case SUBSCRIBE:
case UNSUBSCRIBE:
case SUBACK:
case UNSUBACK:
case PUBACK:
case PUBREC:
case PUBCOMP:
case PUBREL:
return decodeMessageIdVariableHeader(buffer);
case PUBLISH:
return decodePublishVariableHeader(buffer, mqttFixedHeader);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
// Empty variable header
return new Result<>(null, 0);
}
return new Result<>(null, 0);
}
private static Result<MqttConnectVariableHeader> decodeConnectionVariableHeader(ByteBuffer buffer) {
final Result<String> protoString = decodeString(buffer);
int numberOfBytesConsumed = protoString.numberOfBytesConsumed;
final byte protocolLevel = buffer.get();
numberOfBytesConsumed += 1;
final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(protoString.value, protocolLevel);
final int b1 = ByteBufferUtil.readUnsignedByte(buffer);
numberOfBytesConsumed += 1;
final Result<Integer> keepAlive = decodeMsbLsb(buffer);
numberOfBytesConsumed += keepAlive.numberOfBytesConsumed;
final boolean hasUserName = (b1 & 0x80) == 0x80;
final boolean hasPassword = (b1 & 0x40) == 0x40;
final boolean willRetain = (b1 & 0x20) == 0x20;
final int willQos = (b1 & 0x18) >> 3;
final boolean willFlag = (b1 & 0x04) == 0x04;
final boolean cleanSession = (b1 & 0x02) == 0x02;
if (mqttVersion == MqttVersion.MQTT_3_1_1) {
final boolean zeroReservedFlag = (b1 & 0x01) == 0x0;
if (!zeroReservedFlag) {
// MQTT v3.1.1: The Server MUST validate that the reserved flag in the CONNECT Control Packet is
// set to zero and disconnect the Client if it is not zero.
// See http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc385349230
throw new DecoderException("non-zero reserved flag");
}
}
final MqttConnectVariableHeader mqttConnectVariableHeader = new MqttConnectVariableHeader(
mqttVersion.protocolName(),
mqttVersion.protocolLevel(),
hasUserName,
hasPassword,
willRetain,
willQos,
willFlag,
cleanSession,
keepAlive.value);
return new Result<>(mqttConnectVariableHeader, numberOfBytesConsumed);
}
private static Result<MqttConnAckVariableHeader> decodeConnAckVariableHeader(ByteBuffer buffer) {
final boolean sessionPresent = (ByteBufferUtil.readUnsignedByte(buffer) & 0x01) == 0x01;
byte returnCode = ByteBufferUtil.readByte(buffer);
final int numberOfBytesConsumed = 2;
final MqttConnAckVariableHeader mqttConnAckVariableHeader =
new MqttConnAckVariableHeader(MqttConnectReturnCode.valueOf(returnCode), sessionPresent);
return new Result<>(mqttConnAckVariableHeader, numberOfBytesConsumed);
}
private static Result<MqttMessageIdVariableHeader> decodeMessageIdVariableHeader(ByteBuffer buffer) {
final Result<Integer> messageId = decodeMessageId(buffer);
return new Result<>(MqttMessageIdVariableHeader.from(messageId.value),
messageId.numberOfBytesConsumed);
}
private static Result<MqttPublishVariableHeader> decodePublishVariableHeader(
ByteBuffer buffer,
MqttFixedHeader mqttFixedHeader) {
final Result<String> decodedTopic = decodeString(buffer);
if (!MqttCodecUtil.isValidPublishTopicName(decodedTopic.value)) {
throw new DecoderException("invalid publish topic name: " + decodedTopic.value + " (contains wildcards)");
}
int numberOfBytesConsumed = decodedTopic.numberOfBytesConsumed;
int messageId = -1;
if (mqttFixedHeader.qosLevel().value() > 0) {
final Result<Integer> decodedMessageId = decodeMessageId(buffer);
messageId = decodedMessageId.value;
numberOfBytesConsumed += decodedMessageId.numberOfBytesConsumed;
}
final MqttPublishVariableHeader mqttPublishVariableHeader =
new MqttPublishVariableHeader(decodedTopic.value, messageId);
return new Result<>(mqttPublishVariableHeader, numberOfBytesConsumed);
}
private static Result<Integer> decodeMessageId(ByteBuffer buffer) {
final Result<Integer> messageId = decodeMsbLsb(buffer);
if (!MqttCodecUtil.isValidMessageId(messageId.value)) {
throw new DecoderException("invalid messageId: " + messageId.value);
}
return messageId;
}
/**
* Decodes the payload.
*
* @param buffer the buffer to decode from
* @param messageType type of the message being decoded
* @param bytesRemainingInVariablePart bytes remaining
* @param variableHeader variable header of the same message
* @return the payload
*/
private static Result<?> decodePayload(
ByteBuffer buffer,
MqttMessageType messageType,
int bytesRemainingInVariablePart,
Object variableHeader) {
switch (messageType) {
case CONNECT:
return decodeConnectionPayload(buffer, (MqttConnectVariableHeader) variableHeader);
case SUBSCRIBE:
return decodeSubscribePayload(buffer, bytesRemainingInVariablePart);
case SUBACK:
return decodeSubackPayload(buffer, bytesRemainingInVariablePart);
case UNSUBSCRIBE:
return decodeUnsubscribePayload(buffer, bytesRemainingInVariablePart);
case PUBLISH:
return decodePublishPayload(buffer, bytesRemainingInVariablePart);
default:
// unknown payload , no byte consumed
return new Result<>(null, 0);
}
}
private static Result<MqttConnectPayload> decodeConnectionPayload(
ByteBuffer buffer,
MqttConnectVariableHeader mqttConnectVariableHeader) {
final Result<String> decodedClientId = decodeString(buffer);
final String decodedClientIdValue = decodedClientId.value;
final MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(mqttConnectVariableHeader.name(),
(byte) mqttConnectVariableHeader.version());
if (!MqttCodecUtil.isValidClientId(mqttVersion, decodedClientIdValue)) {
throw new MqttIdentifierRejectedException("invalid clientIdentifier: " + decodedClientIdValue);
}
int numberOfBytesConsumed = decodedClientId.numberOfBytesConsumed;
Result<String> decodedWillTopic = null;
Result<byte[]> decodedWillMessage = null;
if (mqttConnectVariableHeader.isWillFlag()) {
decodedWillTopic = decodeString(buffer, 0, 32767);
numberOfBytesConsumed += decodedWillTopic.numberOfBytesConsumed;
decodedWillMessage = decodeByteArray(buffer);
numberOfBytesConsumed += decodedWillMessage.numberOfBytesConsumed;
}
Result<String> decodedUserName = null;
Result<byte[]> decodedPassword = null;
if (mqttConnectVariableHeader.hasUserName()) {
decodedUserName = decodeString(buffer);
numberOfBytesConsumed += decodedUserName.numberOfBytesConsumed;
}
if (mqttConnectVariableHeader.hasPassword()) {
decodedPassword = decodeByteArray(buffer);
numberOfBytesConsumed += decodedPassword.numberOfBytesConsumed;
}
final MqttConnectPayload mqttConnectPayload =
new MqttConnectPayload(
decodedClientId.value,
decodedWillTopic != null ? decodedWillTopic.value : null,
decodedWillMessage != null ? decodedWillMessage.value : null,
decodedUserName != null ? decodedUserName.value : null,
decodedPassword != null ? decodedPassword.value : null);
return new Result<>(mqttConnectPayload, numberOfBytesConsumed);
}
private static Result<MqttSubscribePayload> decodeSubscribePayload(
ByteBuffer buffer,
int bytesRemainingInVariablePart) {
final List<MqttTopicSubscription> subscribeTopics = new ArrayList<>();
int numberOfBytesConsumed = 0;
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
final Result<String> decodedTopicName = decodeString(buffer);
numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
int qos = ByteBufferUtil.readUnsignedByte(buffer) & 0x03;
numberOfBytesConsumed++;
subscribeTopics.add(new MqttTopicSubscription(decodedTopicName.value, MqttQoS.valueOf(qos)));
}
return new Result<>(new MqttSubscribePayload(subscribeTopics), numberOfBytesConsumed);
}
private static Result<MqttSubAckPayload> decodeSubackPayload(
ByteBuffer buffer,
int bytesRemainingInVariablePart) {
final List<Integer> grantedQos = new ArrayList<>();
int numberOfBytesConsumed = 0;
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
int qos = ByteBufferUtil.readUnsignedByte(buffer);
if (qos != MqttQoS.FAILURE.value()) {
qos &= 0x03;
}
numberOfBytesConsumed++;
grantedQos.add(qos);
}
return new Result<>(new MqttSubAckPayload(grantedQos), numberOfBytesConsumed);
}
private static Result<MqttUnsubscribePayload> decodeUnsubscribePayload(
ByteBuffer buffer,
int bytesRemainingInVariablePart) {
final List<String> unsubscribeTopics = new ArrayList<>();
int numberOfBytesConsumed = 0;
while (numberOfBytesConsumed < bytesRemainingInVariablePart) {
final Result<String> decodedTopicName = decodeString(buffer);
numberOfBytesConsumed += decodedTopicName.numberOfBytesConsumed;
unsubscribeTopics.add(decodedTopicName.value);
}
return new Result<>(new MqttUnsubscribePayload(unsubscribeTopics), numberOfBytesConsumed);
}
private static Result<ByteBuffer> decodePublishPayload(ByteBuffer buffer, int bytesRemainingInVariablePart) {
byte[] slice = new byte[bytesRemainingInVariablePart];
buffer.get(slice, 0, bytesRemainingInVariablePart);
ByteBuffer byteBuffer = ByteBuffer.wrap(slice);
return new Result<>(byteBuffer, bytesRemainingInVariablePart);
}
private static Result<String> decodeString(ByteBuffer buffer) {
return decodeString(buffer, 0, Integer.MAX_VALUE);
}
private static Result<String> decodeString(ByteBuffer buffer, int minBytes, int maxBytes) {
final Result<Integer> decodedSize = decodeMsbLsb(buffer);
int size = decodedSize.value;
int numberOfBytesConsumed = decodedSize.numberOfBytesConsumed;
if (size < minBytes || size > maxBytes) {
ByteBufferUtil.skipBytes(buffer, size);
numberOfBytesConsumed += size;
return new Result<>(null, numberOfBytesConsumed);
}
String s = new String(buffer.array(), buffer.position(), size, StandardCharsets.UTF_8);
ByteBufferUtil.skipBytes(buffer, size);
numberOfBytesConsumed += size;
return new Result<>(s, numberOfBytesConsumed);
}
private static Result<byte[]> decodeByteArray(ByteBuffer buffer) {
final Result<Integer> decodedSize = decodeMsbLsb(buffer);
int size = decodedSize.value;
byte[] bytes = new byte[size];
buffer.get(bytes);
// buffer.readBytes(bytes);
return new Result<>(bytes, decodedSize.numberOfBytesConsumed + size);
}
private static Result<Integer> decodeMsbLsb(ByteBuffer buffer) {
return decodeMsbLsb(buffer, 0, 65535);
}
private static Result<Integer> decodeMsbLsb(ByteBuffer buffer, int min, int max) {
short msbSize = ByteBufferUtil.readUnsignedByte(buffer);
short lsbSize = ByteBufferUtil.readUnsignedByte(buffer);
final int numberOfBytesConsumed = 2;
int result = msbSize << 8 | lsbSize;
if (result < min || result > max) {
result = -1;
}
return new Result<>(result, numberOfBytesConsumed);
}
private static final class Result<T> {
private final T value;
private final int numberOfBytesConsumed;
Result(T value, int numberOfBytesConsumed) {
this.value = value;
this.numberOfBytesConsumed = numberOfBytesConsumed;
}
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
/**
* Encodes Mqtt messages into bytes following the protocol specification v3.1
* as described here <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html">MQTTV3.1</a>
*/
public final class MqttEncoder {
public static final MqttEncoder INSTANCE = new MqttEncoder();
/**
* This is the main encoding method.
* It's only visible for testing.
*
* @param message MQTT message to encode
* @return ByteBuffer with encoded bytes
*/
public ByteBuffer doEncode(MqttMessage message) {
switch (message.fixedHeader().messageType()) {
case CONNECT:
return encodeConnectMessage((MqttConnectMessage) message);
case CONNACK:
return encodeConnAckMessage((MqttConnAckMessage) message);
case PUBLISH:
return encodePublishMessage((MqttPublishMessage) message);
case SUBSCRIBE:
return encodeSubscribeMessage((MqttSubscribeMessage) message);
case UNSUBSCRIBE:
return encodeUnsubscribeMessage((MqttUnsubscribeMessage) message);
case SUBACK:
return encodeSubAckMessage((MqttSubAckMessage) message);
case UNSUBACK:
case PUBACK:
case PUBREC:
case PUBREL:
case PUBCOMP:
return encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(message);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
return encodeMessageWithOnlySingleByteFixedHeader(message);
default:
throw new IllegalArgumentException(
"Unknown message type: " + message.fixedHeader().messageType().value());
}
}
private static ByteBuffer encodeConnectMessage(MqttConnectMessage message) {
int payloadBufferSize = 0;
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttConnectVariableHeader variableHeader = message.variableHeader();
MqttConnectPayload payload = message.payload();
MqttVersion mqttVersion = MqttVersion.fromProtocolNameAndLevel(variableHeader.name(),
(byte) variableHeader.version());
// as MQTT 3.1 & 3.1.1 spec, If the User Name Flag is set to 0, the Password Flag MUST be set to 0
if (!variableHeader.hasUserName() && variableHeader.hasPassword()) {
throw new DecoderException("Without a username, the password MUST be not set");
}
// Client id
String clientIdentifier = payload.clientIdentifier();
if (!MqttCodecUtil.isValidClientId(mqttVersion, clientIdentifier)) {
throw new MqttIdentifierRejectedException("invalid clientIdentifier: " + clientIdentifier);
}
byte[] clientIdentifierBytes = encodeStringUtf8(clientIdentifier);
payloadBufferSize += 2 + clientIdentifierBytes.length;
// Will topic and message
String willTopic = payload.willTopic();
byte[] willTopicBytes = willTopic != null ? encodeStringUtf8(willTopic) : ByteBufferUtil.EMPTY;
byte[] willMessage = payload.willMessageInBytes();
byte[] willMessageBytes = willMessage != null ? willMessage : ByteBufferUtil.EMPTY;
if (variableHeader.isWillFlag()) {
payloadBufferSize += 2 + willTopicBytes.length;
payloadBufferSize += 2 + willMessageBytes.length;
}
String userName = payload.userName();
byte[] userNameBytes = userName != null ? encodeStringUtf8(userName) : ByteBufferUtil.EMPTY;
if (variableHeader.hasUserName()) {
payloadBufferSize += 2 + userNameBytes.length;
}
byte[] password = payload.passwordInBytes();
byte[] passwordBytes = password != null ? password : ByteBufferUtil.EMPTY;
if (variableHeader.hasPassword()) {
payloadBufferSize += 2 + passwordBytes.length;
}
// Fixed header
byte[] protocolNameBytes = mqttVersion.protocolNameBytes();
int variableHeaderBufferSize = 2 + protocolNameBytes.length + 4;
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variablePartSize);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
buf.putShort((short) protocolNameBytes.length);
buf.put(protocolNameBytes);
buf.put((byte) variableHeader.version());
buf.put((byte) getConnVariableHeaderFlag(variableHeader));
buf.putShort((short) variableHeader.keepAliveTimeSeconds());
// Payload
buf.putShort((short) clientIdentifierBytes.length);
buf.put(clientIdentifierBytes, 0, clientIdentifierBytes.length);
if (variableHeader.isWillFlag()) {
buf.putShort((short) willTopicBytes.length);
buf.put(willTopicBytes, 0, willTopicBytes.length);
buf.putShort((short) willMessageBytes.length);
buf.put(willMessageBytes, 0, willMessageBytes.length);
}
if (variableHeader.hasUserName()) {
buf.putShort((short) userNameBytes.length);
buf.put(userNameBytes, 0, userNameBytes.length);
}
if (variableHeader.hasPassword()) {
buf.putShort((short) passwordBytes.length);
buf.put(passwordBytes, 0, passwordBytes.length);
}
return buf;
}
private static int getConnVariableHeaderFlag(MqttConnectVariableHeader variableHeader) {
int flagByte = 0;
if (variableHeader.hasUserName()) {
flagByte |= 0x80;
}
if (variableHeader.hasPassword()) {
flagByte |= 0x40;
}
if (variableHeader.isWillRetain()) {
flagByte |= 0x20;
}
flagByte |= (variableHeader.willQos() & 0x03) << 3;
if (variableHeader.isWillFlag()) {
flagByte |= 0x04;
}
if (variableHeader.isCleanSession()) {
flagByte |= 0x02;
}
return flagByte;
}
private static ByteBuffer encodeConnAckMessage(MqttConnAckMessage message) {
ByteBuffer buf = ByteBuffer.allocate(4);
buf.put((byte) getFixedHeaderByte1(message.fixedHeader()));
buf.put((byte) 2);
buf.put((byte) (message.variableHeader().isSessionPresent() ? 0x01 : 0x00));
buf.put(message.variableHeader().connectReturnCode().byteValue());
return buf;
}
private static ByteBuffer encodeSubscribeMessage(MqttSubscribeMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = 0;
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttSubscribePayload payload = message.payload();
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
String topicName = topic.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
payloadBufferSize += 2 + topicNameBytes.length;
payloadBufferSize += 1;
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variablePartSize);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.putShort((short) messageId);
// Payload
for (MqttTopicSubscription topic : payload.topicSubscriptions()) {
String topicName = topic.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
buf.putShort((short) topicNameBytes.length);
buf.put(topicNameBytes, 0, topicNameBytes.length);
buf.put((byte) topic.qualityOfService().value());
}
return buf;
}
private static ByteBuffer encodeUnsubscribeMessage(MqttUnsubscribeMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = 0;
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = message.variableHeader();
MqttUnsubscribePayload payload = message.payload();
for (String topicName : payload.topics()) {
byte[] topicNameBytes = encodeStringUtf8(topicName);
payloadBufferSize += 2 + topicNameBytes.length;
}
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variablePartSize);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
// Variable Header
int messageId = variableHeader.messageId();
buf.putShort((short) messageId);
// Payload
for (String topicName : payload.topics()) {
byte[] topicNameBytes = encodeStringUtf8(topicName);
buf.putShort((short) topicNameBytes.length);
buf.put(topicNameBytes, 0, topicNameBytes.length);
}
return buf;
}
private static ByteBuffer encodeSubAckMessage(MqttSubAckMessage message) {
int variableHeaderBufferSize = 2;
int payloadBufferSize = message.payload().grantedQoSLevels().size();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variablePartSize);
buf.put((byte) getFixedHeaderByte1(message.fixedHeader()));
writeVariableLengthInt(buf, variablePartSize);
buf.putShort((short) message.variableHeader().messageId());
for (int qos : message.payload().grantedQoSLevels()) {
buf.put((byte) qos);
}
return buf;
}
private static ByteBuffer encodePublishMessage(MqttPublishMessage message) {
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttPublishVariableHeader variableHeader = message.variableHeader();
ByteBuffer payload = message.payload().duplicate();
String topicName = variableHeader.topicName();
byte[] topicNameBytes = encodeStringUtf8(topicName);
int variableHeaderBufferSize = 2 + topicNameBytes.length +
(mqttFixedHeader.qosLevel().value() > 0 ? 2 : 0);
int payloadBufferSize = payload.limit();
int variablePartSize = variableHeaderBufferSize + payloadBufferSize;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variablePartSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variablePartSize);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variablePartSize);
buf.putShort((short) topicNameBytes.length);
buf.put(topicNameBytes);
if (mqttFixedHeader.qosLevel().value() > 0) {
buf.putShort((short) variableHeader.packetId());
}
buf.put(payload.array());
return buf;
}
private static ByteBuffer encodeMessageWithOnlySingleByteFixedHeaderAndMessageId(MqttMessage message) {
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
int msgId = variableHeader.messageId();
// variable part only has a message id
int variableHeaderBufferSize = 2;
int fixedHeaderBufferSize = 1 + getVariableLengthInt(variableHeaderBufferSize);
ByteBuffer buf = ByteBuffer.allocate(fixedHeaderBufferSize + variableHeaderBufferSize);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
writeVariableLengthInt(buf, variableHeaderBufferSize);
buf.putShort((short) msgId);
return buf;
}
private static ByteBuffer encodeMessageWithOnlySingleByteFixedHeader(MqttMessage message) {
MqttFixedHeader mqttFixedHeader = message.fixedHeader();
ByteBuffer buf = ByteBuffer.allocate(2);
buf.put((byte) getFixedHeaderByte1(mqttFixedHeader));
buf.put((byte) 0);
return buf;
}
private static int getFixedHeaderByte1(MqttFixedHeader header) {
int ret = 0;
ret |= header.messageType().value() << 4;
if (header.isDup()) {
ret |= 0x08;
}
ret |= header.qosLevel().value() << 1;
if (header.isRetain()) {
ret |= 0x01;
}
return ret;
}
private static void writeVariableLengthInt(ByteBuffer buffer, int num) {
do {
int digit = num % 128;
num /= 128;
if (num > 0) {
digit |= 0x80;
}
buffer.put((byte) digit);
} while (num > 0);
}
private static int getVariableLengthInt(int num) {
int count = 0;
do {
num /= 128;
count++;
} while (num > 0);
return count;
}
private static byte[] encodeStringUtf8(String s) {
return s.getBytes(StandardCharsets.UTF_8);
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.Objects;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#fixed-header">
* MQTTV3.1/fixed-header</a>
*/
public final class MqttFixedHeader {
private final MqttMessageType messageType;
private final boolean isDup;
private final MqttQoS qosLevel;
private final boolean isRetain;
private final int remainingLength;
public MqttFixedHeader(MqttMessageType messageType,
boolean isDup,
MqttQoS qosLevel,
boolean isRetain,
int remainingLength) {
this.messageType = Objects.requireNonNull(messageType, "messageType is null");
this.isDup = isDup;
this.qosLevel = Objects.requireNonNull(qosLevel, "qosLevel is null");
this.isRetain = isRetain;
this.remainingLength = remainingLength;
}
public MqttMessageType messageType() {
return messageType;
}
public boolean isDup() {
return isDup;
}
public MqttQoS qosLevel() {
return qosLevel;
}
public boolean isRetain() {
return isRetain;
}
public int remainingLength() {
return remainingLength;
}
@Override
public String toString() {
return "MqttFixedHeader{" +
"messageType=" + messageType +
", isDup=" + isDup +
", qosLevel=" + qosLevel +
", isRetain=" + isRetain +
", remainingLength=" + remainingLength +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* A {@link MqttIdentifierRejectedException} which is thrown when a CONNECT request contains invalid client identifier.
*/
public final class MqttIdentifierRejectedException extends DecoderException {
private static final long serialVersionUID = -1323503322689614981L;
/**
* Creates a new instance
*/
public MqttIdentifierRejectedException() {
}
/**
* Creates a new instance
*/
public MqttIdentifierRejectedException(String message, Throwable cause) {
super(message, cause);
}
/**
* Creates a new instance
*/
public MqttIdentifierRejectedException(String message) {
super(message);
}
/**
* Creates a new instance
*/
public MqttIdentifierRejectedException(Throwable cause) {
super(cause);
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import org.tio.core.intf.Packet;
/**
* Base class for all MQTT message types.
*/
public class MqttMessage extends Packet {
private final MqttFixedHeader mqttFixedHeader;
private final Object variableHeader;
private final Object payload;
private final DecoderResult decoderResult;
// Constants for fixed-header only message types with all flags set to 0 (see
// http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Table_2.2_-)
public static final MqttMessage PINGREQ = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGREQ, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public static final MqttMessage PINGRESP = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGRESP, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public static final MqttMessage DISCONNECT = new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false,
MqttQoS.AT_MOST_ONCE, false, 0));
public MqttMessage(MqttFixedHeader mqttFixedHeader) {
this(mqttFixedHeader, null, null);
}
public MqttMessage(MqttFixedHeader mqttFixedHeader, Object variableHeader) {
this(mqttFixedHeader, variableHeader, null);
}
public MqttMessage(MqttFixedHeader mqttFixedHeader, Object variableHeader, Object payload) {
this(mqttFixedHeader, variableHeader, payload, DecoderResult.SUCCESS);
}
public MqttMessage(MqttFixedHeader mqttFixedHeader,
Object variableHeader,
Object payload,
DecoderResult decoderResult) {
this.mqttFixedHeader = mqttFixedHeader;
this.variableHeader = variableHeader;
this.payload = payload;
this.decoderResult = decoderResult;
}
public MqttFixedHeader fixedHeader() {
return mqttFixedHeader;
}
public Object variableHeader() {
return variableHeader;
}
public Object payload() {
return payload;
}
public DecoderResult decoderResult() {
return decoderResult;
}
@Override
public String toString() {
return "MqttMessage{" +
"fixedHeader=" + mqttFixedHeader +
", variableHeader=" + variableHeader +
", payload=" + payload +
'}';
}
}
/*
* Copyright 2017 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
public final class MqttMessageBuilders {
public static final class PublishBuilder {
private String topic;
private boolean retained;
private MqttQoS qos;
private ByteBuffer payload;
private int messageId;
PublishBuilder() {
}
public PublishBuilder topicName(String topic) {
this.topic = topic;
return this;
}
public PublishBuilder retained(boolean retained) {
this.retained = retained;
return this;
}
public PublishBuilder qos(MqttQoS qos) {
this.qos = qos;
return this;
}
public PublishBuilder payload(ByteBuffer payload) {
this.payload = payload;
return this;
}
public PublishBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttPublishMessage build() {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retained, 0);
MqttPublishVariableHeader mqttVariableHeader = new MqttPublishVariableHeader(topic, messageId);
return new MqttPublishMessage(mqttFixedHeader, mqttVariableHeader, ByteBufferUtil.clone(payload));
}
}
public static final class ConnectBuilder {
private MqttVersion version = MqttVersion.MQTT_3_1_1;
private String clientId;
private boolean cleanSession;
private boolean hasUser;
private boolean hasPassword;
private int keepAliveSecs;
private boolean willFlag;
private boolean willRetain;
private MqttQoS willQos = MqttQoS.AT_MOST_ONCE;
private String willTopic;
private byte[] willMessage;
private String username;
private byte[] password;
ConnectBuilder() {
}
public ConnectBuilder protocolVersion(MqttVersion version) {
this.version = version;
return this;
}
public ConnectBuilder clientId(String clientId) {
this.clientId = clientId;
return this;
}
public ConnectBuilder cleanSession(boolean cleanSession) {
this.cleanSession = cleanSession;
return this;
}
public ConnectBuilder keepAlive(int keepAliveSecs) {
this.keepAliveSecs = keepAliveSecs;
return this;
}
public ConnectBuilder willFlag(boolean willFlag) {
this.willFlag = willFlag;
return this;
}
public ConnectBuilder willQoS(MqttQoS willQos) {
this.willQos = willQos;
return this;
}
public ConnectBuilder willTopic(String willTopic) {
this.willTopic = willTopic;
return this;
}
public ConnectBuilder willMessage(byte[] willMessage) {
this.willMessage = willMessage;
return this;
}
public ConnectBuilder willRetain(boolean willRetain) {
this.willRetain = willRetain;
return this;
}
public ConnectBuilder hasUser(boolean value) {
this.hasUser = value;
return this;
}
public ConnectBuilder hasPassword(boolean value) {
this.hasPassword = value;
return this;
}
public ConnectBuilder username(String username) {
this.hasUser = username != null;
this.username = username;
return this;
}
public ConnectBuilder password(byte[] password) {
this.hasPassword = password != null;
this.password = password;
return this;
}
public MqttConnectMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnectVariableHeader mqttConnectVariableHeader =
new MqttConnectVariableHeader(
version.protocolName(),
version.protocolLevel(),
hasUser,
hasPassword,
willRetain,
willQos.value(),
willFlag,
cleanSession,
keepAliveSecs);
MqttConnectPayload mqttConnectPayload =
new MqttConnectPayload(clientId, willTopic, willMessage, username, password);
return new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, mqttConnectPayload);
}
}
public static final class SubscribeBuilder {
private List<MqttTopicSubscription> subscriptions;
private int messageId;
SubscribeBuilder() {
}
public SubscribeBuilder addSubscription(MqttQoS qos, String topic) {
if (subscriptions == null) {
subscriptions = new ArrayList<>(5);
}
subscriptions.add(new MqttTopicSubscription(topic, qos));
return this;
}
public SubscribeBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttSubscribeMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttSubscribePayload mqttSubscribePayload = new MqttSubscribePayload(subscriptions);
return new MqttSubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
}
public static final class UnsubscribeBuilder {
private List<String> topicFilters;
private int messageId;
UnsubscribeBuilder() {
}
public UnsubscribeBuilder addTopicFilter(String topic) {
if (topicFilters == null) {
topicFilters = new ArrayList<String>(5);
}
topicFilters.add(topic);
return this;
}
public UnsubscribeBuilder messageId(int messageId) {
this.messageId = messageId;
return this;
}
public MqttUnsubscribeMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
MqttMessageIdVariableHeader mqttVariableHeader = MqttMessageIdVariableHeader.from(messageId);
MqttUnsubscribePayload mqttSubscribePayload = new MqttUnsubscribePayload(topicFilters);
return new MqttUnsubscribeMessage(mqttFixedHeader, mqttVariableHeader, mqttSubscribePayload);
}
}
public static final class ConnAckBuilder {
private MqttConnectReturnCode returnCode;
private boolean sessionPresent;
ConnAckBuilder() {
}
public ConnAckBuilder returnCode(MqttConnectReturnCode returnCode) {
this.returnCode = returnCode;
return this;
}
public ConnAckBuilder sessionPresent(boolean sessionPresent) {
this.sessionPresent = sessionPresent;
return this;
}
public MqttConnAckMessage build() {
MqttFixedHeader mqttFixedHeader =
new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
MqttConnAckVariableHeader mqttConnAckVariableHeader =
new MqttConnAckVariableHeader(returnCode, sessionPresent);
return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
}
}
public static ConnectBuilder connect() {
return new ConnectBuilder();
}
public static ConnAckBuilder connAck() {
return new ConnAckBuilder();
}
public static PublishBuilder publish() {
return new PublishBuilder();
}
public static SubscribeBuilder subscribe() {
return new SubscribeBuilder();
}
public static UnsubscribeBuilder unsubscribe() {
return new UnsubscribeBuilder();
}
private MqttMessageBuilders() {
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
/**
* Utility class with factory methods to create different types of MQTT messages.
*/
public final class MqttMessageFactory {
public static MqttMessage newMessage(MqttFixedHeader mqttFixedHeader, Object variableHeader, Object payload) {
switch (mqttFixedHeader.messageType()) {
case CONNECT:
return new MqttConnectMessage(
mqttFixedHeader,
(MqttConnectVariableHeader) variableHeader,
(MqttConnectPayload) payload);
case CONNACK:
return new MqttConnAckMessage(mqttFixedHeader, (MqttConnAckVariableHeader) variableHeader);
case SUBSCRIBE:
return new MqttSubscribeMessage(
mqttFixedHeader,
(MqttMessageIdVariableHeader) variableHeader,
(MqttSubscribePayload) payload);
case SUBACK:
return new MqttSubAckMessage(
mqttFixedHeader,
(MqttMessageIdVariableHeader) variableHeader,
(MqttSubAckPayload) payload);
case UNSUBACK:
return new MqttUnsubAckMessage(
mqttFixedHeader,
(MqttMessageIdVariableHeader) variableHeader);
case UNSUBSCRIBE:
return new MqttUnsubscribeMessage(
mqttFixedHeader,
(MqttMessageIdVariableHeader) variableHeader,
(MqttUnsubscribePayload) payload);
case PUBLISH:
return new MqttPublishMessage(
mqttFixedHeader,
(MqttPublishVariableHeader) variableHeader,
(ByteBuffer) payload);
case PUBACK:
return new MqttPubAckMessage(mqttFixedHeader, (MqttMessageIdVariableHeader) variableHeader);
case PUBREC:
case PUBREL:
case PUBCOMP:
return new MqttMessage(mqttFixedHeader, variableHeader);
case PINGREQ:
case PINGRESP:
case DISCONNECT:
return new MqttMessage(mqttFixedHeader);
default:
throw new IllegalArgumentException("unknown message type: " + mqttFixedHeader.messageType());
}
}
public static MqttMessage newInvalidMessage(Throwable cause) {
return newInvalidMessage(null, null, cause);
}
public static MqttMessage newInvalidMessage(MqttFixedHeader mqttFixedHeader,
Object variableHeader,
Throwable cause) {
return new MqttMessage(mqttFixedHeader, variableHeader, null, DecoderResult.failure(cause));
}
private MqttMessageFactory() {
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* Variable Header containing only Message Id
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#msg-id">MQTTV3.1/msg-id</a>
*/
public final class MqttMessageIdVariableHeader {
private final int messageId;
public static MqttMessageIdVariableHeader from(int messageId) {
if (messageId < 1 || messageId > 0xffff) {
throw new IllegalArgumentException("messageId: " + messageId + " (expected: 1 ~ 65535)");
}
return new MqttMessageIdVariableHeader(messageId);
}
private MqttMessageIdVariableHeader(int messageId) {
this.messageId = messageId;
}
public int messageId() {
return messageId;
}
@Override
public String toString() {
return "MqttMessageIdVariableHeader{" +
"messageId=" + messageId +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* MQTT Message Types.
*/
public enum MqttMessageType {
/**
* 连接服务端
*/
CONNECT(1),
/**
* 确认连接请求
*/
CONNACK(2),
/**
* 发布消息
*/
PUBLISH(3),
/**
* 发布确认
*/
PUBACK(4),
/**
* 发布收到(QoS 2,第一步)
*/
PUBREC(5),
/**
* 发布释放(QoS 2,第二步)
*/
PUBREL(6),
/**
* 发布完成(QoS 2,第三步)
*/
PUBCOMP(7),
/**
* 订阅主题
*/
SUBSCRIBE(8),
/**
* 订阅确认
*/
SUBACK(9),
/**
* 取消订阅
*/
UNSUBSCRIBE(10),
/**
* 取消订阅确认
*/
UNSUBACK(11),
/**
* 心跳请求
*/
PINGREQ(12),
/**
* 心跳响应
*/
PINGRESP(13),
/**
* 断开连接
*/
DISCONNECT(14);
private final int value;
MqttMessageType(int value) {
this.value = value;
}
public int value() {
return value;
}
public static MqttMessageType valueOf(int type) {
for (MqttMessageType t : values()) {
if (t.value == type) {
return t;
}
}
throw new IllegalArgumentException("unknown message type: " + type);
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#puback">MQTTV3.1/puback</a>
*/
public final class MqttPubAckMessage extends MqttMessage {
public MqttPubAckMessage(MqttFixedHeader mqttFixedHeader, MqttMessageIdVariableHeader variableHeader) {
super(mqttFixedHeader, variableHeader);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.ByteBuffer;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#publish">MQTTV3.1/publish</a>
*/
public class MqttPublishMessage extends MqttMessage {
public MqttPublishMessage(MqttFixedHeader mqttFixedHeader,
MqttPublishVariableHeader variableHeader,
ByteBuffer payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttPublishVariableHeader variableHeader() {
return (MqttPublishVariableHeader) super.variableHeader();
}
@Override
public ByteBuffer payload() {
return (ByteBuffer) super.payload();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* Variable Header of the {@link MqttPublishMessage}
*/
public final class MqttPublishVariableHeader {
private final String topicName;
private final int packetId;
public MqttPublishVariableHeader(String topicName, int packetId) {
this.topicName = topicName;
this.packetId = packetId;
}
public String topicName() {
return topicName;
}
public int packetId() {
return packetId;
}
@Override
public String toString() {
return "MqttPublishVariableHeader{" +
"topicName='" + topicName + '\'' +
", packetId=" + packetId +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License, version 2.0 (the
* "License"); you may not use this file except in compliance with the License. You may obtain a
* copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software distributed under the License
* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
package net.dreamlu.iot.mqtt.codec;
public enum MqttQoS {
/**
* QoS level 0 至多发送一次,发送即丢弃。没有确认消息,也不知道对方是否收到。
*/
AT_MOST_ONCE(0),
/**
* QoS level 1 都要在可变头部中附加一个16位的消息ID,SUBSCRIBE 和 UNSUBSCRIBE 消息使用 QoS level 1。
*/
AT_LEAST_ONCE(1),
/**
* QoS level 2 仅仅在 PUBLISH 类型消息中出现,要求在可变头部中要附加消息ID。
*/
EXACTLY_ONCE(2),
/**
* 失败
*/
FAILURE(0x80);
private final int value;
MqttQoS(int value) {
this.value = value;
}
public int value() {
return value;
}
public static MqttQoS valueOf(int value) {
for (MqttQoS q : values()) {
if (q.value == value) {
return q;
}
}
throw new IllegalArgumentException("invalid QoS: " + value);
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#suback">MQTTV3.1/suback</a>
*/
public final class MqttSubAckMessage extends MqttMessage {
public MqttSubAckMessage(
MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttSubAckPayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
@Override
public MqttSubAckPayload payload() {
return (MqttSubAckPayload) super.payload();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
/**
* Payload of the {@link MqttSubAckMessage}
*/
public class MqttSubAckPayload {
private final List<Integer> grantedQoSLevels;
public MqttSubAckPayload(int... grantedQoSLevels) {
Objects.requireNonNull(grantedQoSLevels, "grantedQoSLevels is null");
List<Integer> list = new ArrayList<>(grantedQoSLevels.length);
for (int v : grantedQoSLevels) {
list.add(v);
}
this.grantedQoSLevels = Collections.unmodifiableList(list);
}
public MqttSubAckPayload(Iterable<Integer> grantedQoSLevels) {
Objects.requireNonNull(grantedQoSLevels, "grantedQoSLevels is null");
List<Integer> list = new ArrayList<>();
for (Integer v : grantedQoSLevels) {
if (v == null) {
break;
}
list.add(v);
}
this.grantedQoSLevels = Collections.unmodifiableList(list);
}
public List<Integer> grantedQoSLevels() {
return grantedQoSLevels;
}
@Override
public String toString() {
return "MqttSubAckPayload{" +
"grantedQoSLevels=" + grantedQoSLevels +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#subscribe">
* MQTTV3.1/subscribe</a>
*/
public final class MqttSubscribeMessage extends MqttMessage {
public MqttSubscribeMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttSubscribePayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
@Override
public MqttSubscribePayload payload() {
return (MqttSubscribePayload) super.payload();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.Collections;
import java.util.List;
/**
* Payload of the {@link MqttSubscribeMessage}
*/
public final class MqttSubscribePayload {
private final List<MqttTopicSubscription> topicSubscriptions;
public MqttSubscribePayload(List<MqttTopicSubscription> topicSubscriptions) {
this.topicSubscriptions = Collections.unmodifiableList(topicSubscriptions);
}
public List<MqttTopicSubscription> topicSubscriptions() {
return topicSubscriptions;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder("MqttSubscribePayload[");
for (MqttTopicSubscription topicSubscription : topicSubscriptions) {
builder.append(topicSubscription).append(", ");
}
if (!topicSubscriptions.isEmpty()) {
builder.setLength(builder.length() - 2);
}
return builder.append(']').toString();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* Contains a topic name and Qos Level.
* This is part of the {@link MqttSubscribePayload}
*/
public final class MqttTopicSubscription {
private final String topicFilter;
private final MqttQoS qualityOfService;
public MqttTopicSubscription(String topicFilter, MqttQoS qualityOfService) {
this.topicFilter = topicFilter;
this.qualityOfService = qualityOfService;
}
public String topicName() {
return topicFilter;
}
public MqttQoS qualityOfService() {
return qualityOfService;
}
@Override
public String toString() {
return "MqttTopicSubscription{" +
"topicFilter='" + topicFilter + '\'' +
", qualityOfService=" + qualityOfService +
'}';
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* A {@link MqttUnacceptableProtocolVersionException} which is thrown when
* a CONNECT request contains unacceptable protocol version.
*/
public final class MqttUnacceptableProtocolVersionException extends DecoderException {
private static final long serialVersionUID = 4914652213232455749L;
/**
* Creates a new instance
*/
public MqttUnacceptableProtocolVersionException() {
}
/**
* Creates a new instance
*/
public MqttUnacceptableProtocolVersionException(String message, Throwable cause) {
super(message, cause);
}
/**
* Creates a new instance
*/
public MqttUnacceptableProtocolVersionException(String message) {
super(message);
}
/**
* Creates a new instance
*/
public MqttUnacceptableProtocolVersionException(Throwable cause) {
super(cause);
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsuback">MQTTV3.1/unsuback</a>
*/
public final class MqttUnsubAckMessage extends MqttMessage {
public MqttUnsubAckMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader) {
super(mqttFixedHeader, variableHeader, null);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
/**
* See <a href="http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html#unsubscribe">
* MQTTV3.1/unsubscribe</a>
*/
public final class MqttUnsubscribeMessage extends MqttMessage {
public MqttUnsubscribeMessage(MqttFixedHeader mqttFixedHeader,
MqttMessageIdVariableHeader variableHeader,
MqttUnsubscribePayload payload) {
super(mqttFixedHeader, variableHeader, payload);
}
@Override
public MqttMessageIdVariableHeader variableHeader() {
return (MqttMessageIdVariableHeader) super.variableHeader();
}
@Override
public MqttUnsubscribePayload payload() {
return (MqttUnsubscribePayload) super.payload();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.util.Collections;
import java.util.List;
/**
* Payload of the {@link MqttUnsubscribeMessage}
*/
public final class MqttUnsubscribePayload {
private final List<String> topics;
public MqttUnsubscribePayload(List<String> topics) {
this.topics = Collections.unmodifiableList(topics);
}
public List<String> topics() {
return topics;
}
@Override
public String toString() {
StringBuilder builder = new StringBuilder("MqttUnsubscribePayload[");
for (String topic : topics) {
builder.append("topicName = ").append(topic).append(", ");
}
if (!topics.isEmpty()) {
builder.setLength(builder.length() - 2);
}
return builder.append("}").toString();
}
}
/*
* Copyright 2014 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package net.dreamlu.iot.mqtt.codec;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
/**
* Mqtt version specific constant values used by multiple classes in mqtt-codec.
*/
public enum MqttVersion {
/**
* mqtt 协议版本
*/
MQTT_3_1("MQIsdp", (byte) 3),
MQTT_3_1_1("MQTT", (byte) 4);
private final String name;
private final byte level;
MqttVersion(String protocolName, byte protocolLevel) {
name = Objects.requireNonNull(protocolName, "protocolName");
level = protocolLevel;
}
public String protocolName() {
return name;
}
public byte[] protocolNameBytes() {
return name.getBytes(StandardCharsets.UTF_8);
}
public byte protocolLevel() {
return level;
}
public static MqttVersion fromProtocolNameAndLevel(String protocolName, byte protocolLevel) {
for (MqttVersion mv : values()) {
if (mv.name.equals(protocolName)) {
if (mv.level == protocolLevel) {
return mv;
} else {
throw new MqttUnacceptableProtocolVersionException(protocolName + " and " + protocolLevel + " are not match");
}
}
}
throw new MqttUnacceptableProtocolVersionException(protocolName + "is unknown protocol name");
}
}
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>mica-mqtt-core</artifactId>
<dependencies>
<dependency>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt-codec</artifactId>
</dependency>
</dependencies>
</project>
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.client;
import net.dreamlu.iot.mqtt.codec.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.client.intf.ClientAioHandler;
import org.tio.core.ChannelContext;
import org.tio.core.TioConfig;
import org.tio.core.exception.AioDecodeException;
import org.tio.core.intf.Packet;
import org.tio.server.AcceptCompletionHandler;
import java.nio.ByteBuffer;
/**
* mqtt 客户端处理
*
* @author L.cm
*/
public class MqttClientAioHandler implements ClientAioHandler {
private static final Logger log = LoggerFactory.getLogger(AcceptCompletionHandler.class);
private final MqttDecoder mqttDecoder;
private final MqttEncoder mqttEncoder;
private final MqttClientProcessor processor;
public MqttClientAioHandler(MqttClientProcessor processor) {
this.mqttDecoder = MqttDecoder.INSTANCE;
this.mqttEncoder = MqttEncoder.INSTANCE;
this.processor =processor;
}
@Override
public Packet heartbeatPacket(ChannelContext channelContext) {
return MqttMessage.PINGREQ;
}
@Override
public Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext) throws AioDecodeException {
return mqttDecoder.decode(buffer);
}
@Override
public ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext channelContext) {
return mqttEncoder.doEncode((MqttMessage) packet);
}
@Override
public void handler(Packet packet, ChannelContext context) throws Exception {
MqttMessage message = (MqttMessage) packet;
// 1. 先判断 mqtt 消息解析是否正常
DecoderResult decoderResult = message.decoderResult();
if (decoderResult.isFailure()) {
processFailure(context, message);
return;
}
MqttFixedHeader fixedHeader = message.fixedHeader();
// 根据消息类型处理消息
MqttMessageType messageType = fixedHeader.messageType();
switch (messageType) {
case CONNACK:
processor.processConAck(context, (MqttConnAckMessage) message);
break;
case SUBACK:
processor.processSubAck((MqttSubAckMessage) message);
break;
case PUBLISH:
processor.processPublish(context, (MqttPublishMessage) message);
break;
case UNSUBACK:
processor.processUnSubAck((MqttUnsubAckMessage) message);
break;
case PUBACK:
processor.processPubAck((MqttPubAckMessage) message);
break;
case PUBREC:
processor.processPubRec(context, message);
break;
case PUBREL:
processor.processPubRel(context, message);
break;
case PUBCOMP:
processor.processPubComp(message);
break;
default:
break;
}
}
/**
* 处理失败
*
* @param context ChannelContext
* @param mqttMessage MqttMessage
*/
private void processFailure(ChannelContext context, MqttMessage mqttMessage) {
// 客户端失败,我认为日志记录异常就行了
Throwable cause = mqttMessage.decoderResult().getCause();
log.error(cause.getMessage(), cause);
}
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.client;
import lombok.RequiredArgsConstructor;
import net.dreamlu.iot.mqtt.codec.MqttConnectMessage;
import net.dreamlu.iot.mqtt.codec.MqttMessageBuilders;
import org.tio.client.DefaultClientAioListener;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import java.nio.charset.StandardCharsets;
/**
* mqtt 客户端监听器
*
* @author L.cm
*/
@RequiredArgsConstructor
public class MqttClientAioListener extends DefaultClientAioListener {
private final String clientId;
private final String username;
private final String password;
@Override
public void onAfterConnected(ChannelContext context, boolean isConnected, boolean isReconnect) {
if (isConnected) {
// 1. 建立连接后发送 mqtt 连接的消息
MqttConnectMessage message = MqttMessageBuilders.connect()
.clientId(clientId)
.username(username)
.password(password.getBytes(StandardCharsets.UTF_8))
.build();
Tio.send(context, message);
}
}
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.client;
import net.dreamlu.iot.mqtt.codec.*;
import org.tio.core.ChannelContext;
/**
* mqtt 客户端消息处理器
*
* @author L.cm
*/
public interface MqttClientProcessor {
/**
* 处理服务端链接 ack
*
* @param context ChannelContext
* @param message MqttConnAckMessage
*/
void processConAck(ChannelContext context, MqttConnAckMessage message);
/**
* 处理服务端订阅的 ack
*
* @param message MqttSubAckMessage
*/
void processSubAck(MqttSubAckMessage message);
/**
* 处理服务端 publish 的消息
*
* @param context ChannelContext
* @param message MqttPublishMessage
*/
void processPublish(ChannelContext context, MqttPublishMessage message);
/**
* 处理服务端解除订阅的 ack
*
* @param message MqttSubAckMessage
*/
void processUnSubAck(MqttUnsubAckMessage message);
/**
* 处理服务端 publish 的 ack
*
* @param message MqttPubAckMessage
*/
void processPubAck(MqttPubAckMessage message);
/**
* 处理服务端 publish rec
*
* @param context ChannelContext
* @param message MqttPubAckMessage
*/
void processPubRec(ChannelContext context, MqttMessage message);
/**
* 处理服务端 publish rel
*
* @param context ChannelContext
* @param message MqttPubAckMessage
*/
void processPubRel(ChannelContext context, MqttMessage message);
/**
* 处理服务端 publish comp
*
* @param message MqttPubAckMessage
*/
void processPubComp(MqttMessage message);
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.server;
import net.dreamlu.iot.mqtt.codec.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.core.TioConfig;
import org.tio.core.exception.AioDecodeException;
import org.tio.core.intf.Packet;
import org.tio.server.AcceptCompletionHandler;
import org.tio.server.intf.ServerAioHandler;
import java.nio.ByteBuffer;
/**
* @author L.cm
*/
public class MqttServerAioHandler implements ServerAioHandler {
private static final Logger log = LoggerFactory.getLogger(AcceptCompletionHandler.class);
private final MqttDecoder mqttDecoder;
private final MqttEncoder mqttEncoder;
private final MqttServerProcessor processor;
public MqttServerAioHandler(MqttServerProcessor processor) {
this.mqttDecoder = MqttDecoder.INSTANCE;
this.mqttEncoder = MqttEncoder.INSTANCE;
this.processor = processor;
}
/**
* 根据ByteBuffer解码成业务需要的Packet对象.
* 如果收到的数据不全,导致解码失败,请返回null,在下次消息来时框架层会自动续上前面的收到的数据
*
* @param buffer 参与本次希望解码的ByteBuffer
* @param limit ByteBuffer的limit
* @param position ByteBuffer的position,不一定是0哦
* @param readableLength ByteBuffer参与本次解码的有效数据(= limit - position)
* @param channelContext ChannelContext
* @return Packet
* @throws AioDecodeException AioDecodeException
*/
@Override
public Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext) throws AioDecodeException {
return mqttDecoder.decode(buffer);
}
/**
* 编码
*
* @param packet Packet
* @param tioConfig TioConfig
* @param channelContext ChannelContext
* @return ByteBuffer
*/
@Override
public ByteBuffer encode(Packet packet, TioConfig tioConfig, ChannelContext channelContext) {
return mqttEncoder.doEncode((MqttMessage) packet);
}
/**
* 处理消息包
*
* @param packet Packet
* @param context ChannelContext
* @throws Exception Exception
*/
@Override
public void handler(Packet packet, ChannelContext context) throws Exception {
MqttMessage mqttMessage = (MqttMessage) packet;
// 1. 先判断 mqtt 消息解析是否正常
DecoderResult decoderResult = mqttMessage.decoderResult();
if (decoderResult.isFailure()) {
processFailure(context, mqttMessage);
return;
}
MqttFixedHeader fixedHeader = mqttMessage.fixedHeader();
MqttMessageType messageType = fixedHeader.messageType();
log.debug("MqttMessageType:{}", messageType);
switch (messageType) {
case CONNECT:
processor.processConnect(context, (MqttConnectMessage) mqttMessage);
break;
case CONNACK:
break;
case PUBLISH:
processor.processPublish(context, (MqttPublishMessage) mqttMessage);
break;
case PUBACK:
processor.processPubAck(context, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
break;
case PUBREC:
processor.processPubRec(context, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
break;
case PUBREL:
processor.processPubRel(context, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
break;
case PUBCOMP:
processor.processPubComp(context, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
break;
case SUBSCRIBE:
processor.processSubscribe(context, (MqttSubscribeMessage) mqttMessage);
break;
case SUBACK:
break;
case UNSUBSCRIBE:
processor.processUnSubscribe(context, (MqttUnsubscribeMessage) mqttMessage);
break;
case UNSUBACK:
break;
case PINGREQ:
processor.processPingReq(context);
break;
case PINGRESP:
break;
case DISCONNECT:
processor.processDisConnect(context);
break;
default:
break;
}
}
/**
* 处理失败
*
* @param context ChannelContext
* @param mqttMessage MqttMessage
*/
private void processFailure(ChannelContext context, MqttMessage mqttMessage) {
Throwable cause = mqttMessage.decoderResult().getCause();
if (cause instanceof MqttUnacceptableProtocolVersionException) {
log.error(cause.getMessage());
// 不支持的协议版本
MqttConnAckMessage message = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION)
.sessionPresent(false)
.build();
Tio.send(context, message);
} else if (cause instanceof MqttIdentifierRejectedException) {
log.error(cause.getMessage());
// 不合格的 clientId
MqttConnAckMessage message = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED)
.sessionPresent(false)
.build();
Tio.send(context, message);
} else {
log.error(cause.getMessage(), cause);
// 发送断开连接,是否强制关闭客户端连接???
Tio.send(context, MqttMessage.DISCONNECT);
}
}
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.server;
import org.tio.core.ChannelContext;
import org.tio.core.DefaultAioListener;
/**
* mqtt 服务监听
*
* @author L.cm
*/
public class MqttServerAioListener extends DefaultAioListener {
@Override
public boolean onHeartbeatTimeout(ChannelContext channelContext, Long interval, int heartbeatTimeoutCount) {
return true;
}
}
/*
* Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.dreamlu.iot.mqtt.core.server;
import net.dreamlu.iot.mqtt.codec.*;
import org.tio.core.ChannelContext;
/**
* mqtt broker 处理器
*
* @author L.cm
*/
public interface MqttServerProcessor {
/**
* 处理链接
*
* @param context ChannelContext
* @param message MqttConnectMessage
*/
void processConnect(ChannelContext context, MqttConnectMessage message);
/**
* Publish
*
* @param context ChannelContext
* @param message MqttPublishMessage
*/
void processPublish(ChannelContext context, MqttPublishMessage message);
/**
* PubAck
*
* @param context ChannelContext
* @param variableHeader MqttMessageIdVariableHeader
*/
void processPubAck(ChannelContext context, MqttMessageIdVariableHeader variableHeader);
/**
* PubRec
*
* @param context ChannelContext
* @param variableHeader MqttMessageIdVariableHeader
*/
void processPubRec(ChannelContext context, MqttMessageIdVariableHeader variableHeader);
/**
* PubRel
*
* @param context ChannelContext
* @param variableHeader MqttMessageIdVariableHeader
*/
void processPubRel(ChannelContext context, MqttMessageIdVariableHeader variableHeader);
/**
* PubComp
*
* @param context ChannelContext
* @param variableHeader MqttMessageIdVariableHeader
*/
void processPubComp(ChannelContext context, MqttMessageIdVariableHeader variableHeader);
/**
* 监听
*
* @param context ChannelContext
* @param message MqttSubscribeMessage
*/
void processSubscribe(ChannelContext context, MqttSubscribeMessage message);
/**
* 取消监听
*
* @param context ChannelContext
* @param message MqttUnsubscribeMessage
*/
void processUnSubscribe(ChannelContext context, MqttUnsubscribeMessage message);
/**
* ping 消息处理
*
* @param context ChannelContext
*/
void processPingReq(ChannelContext context);
/**
* 断开连接
*
* @param context ChannelContext
*/
void processDisConnect(ChannelContext context);
}
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt</artifactId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<artifactId>mica-mqtt-example</artifactId>
<dependencies>
<dependency>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt-core</artifactId>
</dependency>
<!--用于与slf4j保持桥接-->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.13.3</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<archive>
<manifest>
<addClasspath>true</addClasspath>
<mainClass>net.dreamlu.iot.mqtt.example.MqttServerTest</mainClass>
</manifest>
</archive>
</configuration>
</plugin>
</plugins>
</build>
</project>
package net.dreamlu.iot.mqtt.client;
import net.dreamlu.iot.mqtt.codec.*;
import net.dreamlu.iot.mqtt.core.client.MqttClientProcessor;
import org.tio.core.ChannelContext;
import java.nio.ByteBuffer;
/**
* 示例客户端处理
*
* @author L.cm
*/
public class MqttClientProcessorImpl implements MqttClientProcessor {
@Override
public void processConAck(ChannelContext context, MqttConnAckMessage message) {
System.out.println(message);
}
@Override
public void processSubAck(MqttSubAckMessage message) {
System.out.println(message);
}
@Override
public void processPublish(ChannelContext context, MqttPublishMessage message) {
ByteBuffer byteBuffer = message.payload();
if (byteBuffer != null) {
System.out.println(ByteBufferUtil.toString(byteBuffer));
}
}
@Override
public void processUnSubAck(MqttUnsubAckMessage message) {
System.out.println(message);
}
@Override
public void processPubAck(MqttPubAckMessage message) {
System.out.println(message);
}
@Override
public void processPubRec(ChannelContext context, MqttMessage message) {
System.out.println(message);
}
@Override
public void processPubRel(ChannelContext context, MqttMessage message) {
System.out.println(message);
}
@Override
public void processPubComp(MqttMessage message) {
System.out.println(message);
}
}
package net.dreamlu.iot.mqtt.client;
import net.dreamlu.iot.mqtt.codec.*;
import net.dreamlu.iot.mqtt.core.client.MqttClientAioHandler;
import net.dreamlu.iot.mqtt.core.client.MqttClientAioListener;
import net.dreamlu.iot.mqtt.core.client.MqttClientProcessor;
import org.tio.client.ClientChannelContext;
import org.tio.client.ClientTioConfig;
import org.tio.client.ReconnConf;
import org.tio.client.TioClient;
import org.tio.client.intf.ClientAioHandler;
import org.tio.client.intf.ClientAioListener;
import org.tio.core.Node;
import org.tio.core.Tio;
import java.nio.ByteBuffer;
import java.util.Timer;
import java.util.TimerTask;
/**
* 客户端测试
*
* @author L.cm
*/
public class MqttClientTest {
public static void main(String[] args) throws Exception {
MqttClientProcessor processor = new MqttClientProcessorImpl();
ClientAioHandler clientAioHandler = new MqttClientAioHandler(processor);
ClientAioListener clientAioListener = new MqttClientAioListener("MqttClientTest", "admin", "123456");
ReconnConf reconnConf = new ReconnConf();
ClientTioConfig tioConfig = new ClientTioConfig(clientAioHandler, clientAioListener, reconnConf);
TioClient tioClient = new TioClient(tioConfig);
ClientChannelContext context = tioClient.connect(new Node("127.0.0.1", 1883), 1000);
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
if (!context.isClosed) {
MqttPublishMessage message = (MqttPublishMessage) MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
new MqttPublishVariableHeader("testtopicxx", 0), ByteBuffer.wrap("mica最牛皮".getBytes()));
Tio.send(context, message);
}
}
}, 1000, 2000);
}
}
package net.dreamlu.iot.mqtt.example;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.codec.*;
import net.dreamlu.iot.mqtt.core.server.MqttServerProcessor;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.utils.hutool.StrUtil;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.Collectors;
/**
* mqtt broker 处理器
*
* @author L.cm
*/
@Slf4j
@RequiredArgsConstructor
public class MqttBrokerProcessorImpl implements MqttServerProcessor {
private static final String MQTT_CLIENT_ID_KEY = "mqttClientId";
@Override
public void processConnect(ChannelContext context, MqttConnectMessage mqttMessage) {
MqttConnectPayload payload = mqttMessage.payload();
String clientId = payload.clientIdentifier();
// 1. 客户端必须提供clientId, 不管cleanSession是否为1, 此处没有参考标准协议实现
if (StrUtil.isBlank(clientId)) {
refusedIdentifierRejected(context);
return;
}
log.debug("CONNECT - clientId: {}", clientId);
// 2. 认证
String userName = payload.userName();
String password = payload.password();
boolean authResult = false;
if (false) {
MqttConnAckMessage message = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD)
.sessionPresent(false)
.build();
Tio.send(context, message);
return;
}
// 3. 设置 clientId
context.set(MQTT_CLIENT_ID_KEY, clientId);
// 4. 返回 ack
MqttMessage message = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_ACCEPTED)
.sessionPresent(false)
.build();
Tio.send(context, message);
}
private void refusedIdentifierRejected(ChannelContext context) {
MqttConnAckMessage message = MqttMessageBuilders.connAck()
.returnCode(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED)
.sessionPresent(false)
.build();
Tio.send(context, message);
}
@Override
public void processPublish(ChannelContext context, MqttPublishMessage message) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PUBLISH - clientId: {}", clientId);
MqttFixedHeader fixedHeader = message.fixedHeader();
ByteBuffer payload = message.payload();
if (payload != null) {
System.out.println(ByteBufferUtil.toString(payload));
}
}
@Override
public void processPubAck(ChannelContext context, MqttMessageIdVariableHeader variableHeader) {
int messageId = variableHeader.messageId();
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PUBACK - clientId: {}, messageId: {}", clientId, messageId);
}
@Override
public void processPubRec(ChannelContext context, MqttMessageIdVariableHeader variableHeader) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PUBREC - clientId: {}, messageId: {}", clientId, variableHeader.messageId());
MqttMessage message = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_MOST_ONCE, false, 0),
MqttMessageIdVariableHeader.from(variableHeader.messageId()), null);
Tio.send(context, message);
}
@Override
public void processPubRel(ChannelContext context, MqttMessageIdVariableHeader variableHeader) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PUBREL - clientId: {}, messageId: {}", clientId, variableHeader.messageId());
MqttMessage message = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0),
MqttMessageIdVariableHeader.from(variableHeader.messageId()), null);
Tio.send(context, message);
}
@Override
public void processPubComp(ChannelContext context, MqttMessageIdVariableHeader variableHeader) {
int messageId = variableHeader.messageId();
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PUBCOMP - clientId: {}, messageId: {}", clientId, messageId);
}
@Override
public void processSubscribe(ChannelContext context, MqttSubscribeMessage message) {
List<MqttTopicSubscription> topicSubscriptions = message.payload().topicSubscriptions();
List<Integer> mqttQoSList = topicSubscriptions.stream()
.map(MqttTopicSubscription::qualityOfService)
.map(MqttQoS::value)
.collect(Collectors.toList());
MqttMessage subAckMessage = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
MqttMessageIdVariableHeader.from(message.variableHeader().messageId()),
new MqttSubAckPayload(mqttQoSList));
Tio.send(context, subAckMessage);
}
@Override
public void processUnSubscribe(ChannelContext context, MqttUnsubscribeMessage mqttMessage) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("UnSubscribe - clientId: {}", clientId);
MqttMessage message = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
MqttMessageIdVariableHeader.from(mqttMessage.variableHeader().messageId()), null);
Tio.send(context, message);
}
@Override
public void processPingReq(ChannelContext context) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("PINGREQ - clientId: {}", clientId);
MqttMessage message = MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0),
null, null);
Tio.send(context, message);
}
@Override
public void processDisConnect(ChannelContext context) {
String clientId = (String) context.get(MQTT_CLIENT_ID_KEY);
log.debug("DISCONNECT - clientId: {}", clientId);
context.setClosed(true);
}
}
package net.dreamlu.iot.mqtt.example;
import net.dreamlu.iot.mqtt.codec.*;
import net.dreamlu.iot.mqtt.core.server.MqttServerAioHandler;
import net.dreamlu.iot.mqtt.core.server.MqttServerAioListener;
import net.dreamlu.iot.mqtt.core.server.MqttServerProcessor;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.server.ServerTioConfig;
import org.tio.server.TioServer;
import org.tio.server.intf.ServerAioHandler;
import org.tio.server.intf.ServerAioListener;
import org.tio.utils.lock.SetWithLock;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
/**
* mqtt 服务端测试
*
* @author L.cm
*/
public class MqttServerTest {
public static void main(String[] args) throws IOException {
int socketPort = 1883;
MqttServerProcessor brokerHandler = new MqttBrokerProcessorImpl();
// 处理消息
ServerAioHandler handler = new MqttServerAioHandler(brokerHandler);
// 监听
ServerAioListener listener = new MqttServerAioListener();
// 配置
ServerTioConfig config = new ServerTioConfig("mqtt-server", handler, listener);
TioServer tioServer = new TioServer(config);
// 设置timeout
config.setHeartbeatTimeout(500);
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
SetWithLock<ChannelContext> contextSet = Tio.getAll(config);
Set<ChannelContext> channelContexts = contextSet.getObj();
channelContexts.forEach(context -> {
MqttPublishMessage message = (MqttPublishMessage) MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
new MqttPublishVariableHeader("testtopic", 0), ByteBuffer.wrap("mica最牛皮".getBytes()));
Tio.send(context, message);
});
}
}, 1000, 2000);
// 启动
tioServer.start("127.0.0.1", socketPort);
}
}
log4j.rootLogger = DEBUG
appender.console.type = Console
appender.console.name = STDOUT
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d %-5p %c{2} - %m%n
rootLogger.level = debug
rootLogger.appenderRef.stdout.ref = STDOUT
rootLogger.appenderRef.log.ref = log
#!/bin/sh
# ----------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# ----------------------------------------------------------------------------
# ----------------------------------------------------------------------------
# Maven2 Start Up Batch script
#
# Required ENV vars:
# ------------------
# JAVA_HOME - location of a JDK home dir
#
# Optional ENV vars
# -----------------
# M2_HOME - location of maven2's installed home dir
# MAVEN_OPTS - parameters passed to the Java VM when running Maven
# e.g. to debug Maven itself, use
# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
# MAVEN_SKIP_RC - flag to disable loading of mavenrc files
# ----------------------------------------------------------------------------
if [ -z "$MAVEN_SKIP_RC" ] ; then
if [ -f /etc/mavenrc ] ; then
. /etc/mavenrc
fi
if [ -f "$HOME/.mavenrc" ] ; then
. "$HOME/.mavenrc"
fi
fi
# OS specific support. $var _must_ be set to either true or false.
cygwin=false;
darwin=false;
mingw=false
case "`uname`" in
CYGWIN*) cygwin=true ;;
MINGW*) mingw=true;;
Darwin*) darwin=true
# Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
# See https://developer.apple.com/library/mac/qa/qa1170/_index.html
if [ -z "$JAVA_HOME" ]; then
if [ -x "/usr/libexec/java_home" ]; then
export JAVA_HOME="`/usr/libexec/java_home`"
else
export JAVA_HOME="/Library/Java/Home"
fi
fi
;;
esac
if [ -z "$JAVA_HOME" ] ; then
if [ -r /etc/gentoo-release ] ; then
JAVA_HOME=`java-config --jre-home`
fi
fi
if [ -z "$M2_HOME" ] ; then
## resolve links - $0 may be a link to maven's home
PRG="$0"
# need this for relative symlinks
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG="`dirname "$PRG"`/$link"
fi
done
saveddir=`pwd`
M2_HOME=`dirname "$PRG"`/..
# make it fully qualified
M2_HOME=`cd "$M2_HOME" && pwd`
cd "$saveddir"
# echo Using m2 at $M2_HOME
fi
# For Cygwin, ensure paths are in UNIX format before anything is touched
if $cygwin ; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --unix "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
fi
# For Mingw, ensure paths are in UNIX format before anything is touched
if $mingw ; then
[ -n "$M2_HOME" ] &&
M2_HOME="`(cd "$M2_HOME"; pwd)`"
[ -n "$JAVA_HOME" ] &&
JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
fi
if [ -z "$JAVA_HOME" ]; then
javaExecutable="`which javac`"
if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
# readlink(1) is not available as standard on Solaris 10.
readLink=`which readlink`
if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
if $darwin ; then
javaHome="`dirname \"$javaExecutable\"`"
javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
else
javaExecutable="`readlink -f \"$javaExecutable\"`"
fi
javaHome="`dirname \"$javaExecutable\"`"
javaHome=`expr "$javaHome" : '\(.*\)/bin'`
JAVA_HOME="$javaHome"
export JAVA_HOME
fi
fi
fi
if [ -z "$JAVACMD" ] ; then
if [ -n "$JAVA_HOME" ] ; then
if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
# IBM's JDK on AIX uses strange locations for the executables
JAVACMD="$JAVA_HOME/jre/sh/java"
else
JAVACMD="$JAVA_HOME/bin/java"
fi
else
JAVACMD="`which java`"
fi
fi
if [ ! -x "$JAVACMD" ] ; then
echo "Error: JAVA_HOME is not defined correctly." >&2
echo " We cannot execute $JAVACMD" >&2
exit 1
fi
if [ -z "$JAVA_HOME" ] ; then
echo "Warning: JAVA_HOME environment variable is not set."
fi
CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
# traverses directory structure from process work directory to filesystem root
# first directory with .mvn subdirectory is considered project base directory
find_maven_basedir() {
if [ -z "$1" ]
then
echo "Path not specified to find_maven_basedir"
return 1
fi
basedir="$1"
wdir="$1"
while [ "$wdir" != '/' ] ; do
if [ -d "$wdir"/.mvn ] ; then
basedir=$wdir
break
fi
# workaround for JBEAP-8937 (on Solaris 10/Sparc)
if [ -d "${wdir}" ]; then
wdir=`cd "$wdir/.."; pwd`
fi
# end of workaround
done
echo "${basedir}"
}
# concatenates all lines of a file
concat_lines() {
if [ -f "$1" ]; then
echo "$(tr -s '\n' ' ' < "$1")"
fi
}
BASE_DIR=`find_maven_basedir "$(pwd)"`
if [ -z "$BASE_DIR" ]; then
exit 1;
fi
##########################################################################################
# Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
# This allows using the maven wrapper in projects that prohibit checking in binary data.
##########################################################################################
if [ -r "$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" ]; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found .mvn/wrapper/maven-wrapper.jar"
fi
else
if [ "$MVNW_VERBOSE" = true ]; then
echo "Couldn't find .mvn/wrapper/maven-wrapper.jar, downloading it ..."
fi
if [ -n "$MVNW_REPOURL" ]; then
jarUrl="$MVNW_REPOURL/io/takari/maven-wrapper/0.5.5/maven-wrapper-0.5.5.jar"
else
jarUrl="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.5/maven-wrapper-0.5.5.jar"
fi
while IFS="=" read key value; do
case "$key" in (wrapperUrl) jarUrl="$value"; break ;;
esac
done < "$BASE_DIR/.mvn/wrapper/maven-wrapper.properties"
if [ "$MVNW_VERBOSE" = true ]; then
echo "Downloading from: $jarUrl"
fi
wrapperJarPath="$BASE_DIR/.mvn/wrapper/maven-wrapper.jar"
if $cygwin; then
wrapperJarPath=`cygpath --path --windows "$wrapperJarPath"`
fi
if command -v wget > /dev/null; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found wget ... using wget"
fi
if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
wget "$jarUrl" -O "$wrapperJarPath"
else
wget --http-user=$MVNW_USERNAME --http-password=$MVNW_PASSWORD "$jarUrl" -O "$wrapperJarPath"
fi
elif command -v curl > /dev/null; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found curl ... using curl"
fi
if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
curl -o "$wrapperJarPath" "$jarUrl" -f
else
curl --user $MVNW_USERNAME:$MVNW_PASSWORD -o "$wrapperJarPath" "$jarUrl" -f
fi
else
if [ "$MVNW_VERBOSE" = true ]; then
echo "Falling back to using Java to download"
fi
javaClass="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.java"
# For Cygwin, switch paths to Windows format before running javac
if $cygwin; then
javaClass=`cygpath --path --windows "$javaClass"`
fi
if [ -e "$javaClass" ]; then
if [ ! -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
if [ "$MVNW_VERBOSE" = true ]; then
echo " - Compiling MavenWrapperDownloader.java ..."
fi
# Compiling the Java class
("$JAVA_HOME/bin/javac" "$javaClass")
fi
if [ -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
# Running the downloader
if [ "$MVNW_VERBOSE" = true ]; then
echo " - Running MavenWrapperDownloader.java ..."
fi
("$JAVA_HOME/bin/java" -cp .mvn/wrapper MavenWrapperDownloader "$MAVEN_PROJECTBASEDIR")
fi
fi
fi
fi
##########################################################################################
# End of extension
##########################################################################################
export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
if [ "$MVNW_VERBOSE" = true ]; then
echo $MAVEN_PROJECTBASEDIR
fi
MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
# For Cygwin, switch paths to Windows format before running java
if $cygwin; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --path --windows "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
[ -n "$MAVEN_PROJECTBASEDIR" ] &&
MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
fi
# Provide a "standardized" way to retrieve the CLI args that will
# work with both Windows and non-Windows executions.
MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
export MAVEN_CMD_LINE_ARGS
WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
exec "$JAVACMD" \
$MAVEN_OPTS \
-classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
"-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"
@REM ----------------------------------------------------------------------------
@REM Licensed to the Apache Software Foundation (ASF) under one
@REM or more contributor license agreements. See the NOTICE file
@REM distributed with this work for additional information
@REM regarding copyright ownership. The ASF licenses this file
@REM to you under the Apache License, Version 2.0 (the
@REM "License"); you may not use this file except in compliance
@REM with the License. You may obtain a copy of the License at
@REM
@REM http://www.apache.org/licenses/LICENSE-2.0
@REM
@REM Unless required by applicable law or agreed to in writing,
@REM software distributed under the License is distributed on an
@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@REM KIND, either express or implied. See the License for the
@REM specific language governing permissions and limitations
@REM under the License.
@REM ----------------------------------------------------------------------------
@REM ----------------------------------------------------------------------------
@REM Maven2 Start Up Batch script
@REM
@REM Required ENV vars:
@REM JAVA_HOME - location of a JDK home dir
@REM
@REM Optional ENV vars
@REM M2_HOME - location of maven2's installed home dir
@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a key stroke before ending
@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
@REM e.g. to debug Maven itself, use
@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
@REM ----------------------------------------------------------------------------
@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
@echo off
@REM set title of command window
title %0
@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO%
@REM set %HOME% to equivalent of $HOME
if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
@REM Execute a user defined script before this one
if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
@REM check for pre script, once with legacy .bat ending and once with .cmd ending
if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat"
if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd"
:skipRcPre
@setlocal
set ERROR_CODE=0
@REM To isolate internal variables from possible post scripts, we use another setlocal
@setlocal
@REM ==== START VALIDATION ====
if not "%JAVA_HOME%" == "" goto OkJHome
echo.
echo Error: JAVA_HOME not found in your environment. >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
:OkJHome
if exist "%JAVA_HOME%\bin\java.exe" goto init
echo.
echo Error: JAVA_HOME is set to an invalid directory. >&2
echo JAVA_HOME = "%JAVA_HOME%" >&2
echo Please set the JAVA_HOME variable in your environment to match the >&2
echo location of your Java installation. >&2
echo.
goto error
@REM ==== END VALIDATION ====
:init
@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
@REM Fallback to current working directory if not found.
set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
set EXEC_DIR=%CD%
set WDIR=%EXEC_DIR%
:findBaseDir
IF EXIST "%WDIR%"\.mvn goto baseDirFound
cd ..
IF "%WDIR%"=="%CD%" goto baseDirNotFound
set WDIR=%CD%
goto findBaseDir
:baseDirFound
set MAVEN_PROJECTBASEDIR=%WDIR%
cd "%EXEC_DIR%"
goto endDetectBaseDir
:baseDirNotFound
set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
cd "%EXEC_DIR%"
:endDetectBaseDir
IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
@setlocal EnableExtensions EnableDelayedExpansion
for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
:endReadAdditionalConfig
SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.5/maven-wrapper-0.5.5.jar"
FOR /F "tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
)
@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
if exist %WRAPPER_JAR% (
if "%MVNW_VERBOSE%" == "true" (
echo Found %WRAPPER_JAR%
)
) else (
if not "%MVNW_REPOURL%" == "" (
SET DOWNLOAD_URL="%MVNW_REPOURL%/io/takari/maven-wrapper/0.5.5/maven-wrapper-0.5.5.jar"
)
if "%MVNW_VERBOSE%" == "true" (
echo Couldn't find %WRAPPER_JAR%, downloading it ...
echo Downloading from: %DOWNLOAD_URL%
)
powershell -Command "&{"^
"$webclient = new-object System.Net.WebClient;"^
"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
"}"^
"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
"}"
if "%MVNW_VERBOSE%" == "true" (
echo Finished downloading %WRAPPER_JAR%
)
)
@REM End of extension
@REM Provide a "standardized" way to retrieve the CLI args that will
@REM work with both Windows and non-Windows executions.
set MAVEN_CMD_LINE_ARGS=%*
%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
if ERRORLEVEL 1 goto error
goto end
:error
set ERROR_CODE=1
:end
@endlocal & set ERROR_CODE=%ERROR_CODE%
if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost
@REM check for post script, once with legacy .bat ending and once with .cmd ending
if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat"
if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd"
:skipRcPost
@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
if "%MAVEN_BATCH_PAUSE%" == "on" pause
if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE%
exit /B %ERROR_CODE%
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>pom</packaging>
<name>${project.artifactId}</name>
<description>Mica mqtt tools.</description>
<properties>
<java.version>1.8</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<mica-iot.version>0.0.1-SNAPSHOT</mica-iot.version>
<tio.version>3.6.2.v20200808-RELEASE</tio.version>
<lombok.version>1.18.12</lombok.version>
</properties>
<modules>
<module>mica-mqtt-codec</module>
<module>mica-mqtt-core</module>
<module>mica-mqtt-example</module>
</modules>
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.t-io</groupId>
<artifactId>tio-core</artifactId>
<version>${tio.version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt-codec</artifactId>
<version>${mica-iot.version}</version>
</dependency>
<dependency>
<groupId>net.dreamlu</groupId>
<artifactId>mica-mqtt-core</artifactId>
<version>${mica-iot.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
<encoding>${project.build.sourceEncoding}</encoding>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<!--阿里云主仓库,代理了maven central和jcenter仓库-->
<repository>
<id>aliyun</id>
<name>aliyun</name>
<url>https://maven.aliyun.com/repository/public</url>
</repository>
</repositories>
</project>
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册