HarmonyOS 鸿蒙应用开发(十二、paho-mqtt的mqtt客户端库移植)
PAHO MQTT 是 Eclipse Foundation 的一个开源项目，提供了用于 MQTT 协议的客户端库。这个库为使用 MQTT 协议的应用程序提供了丰富的功能和灵活性。轻量级mqtt服务器mosquitto也是其中有名的一个开源项目。
PAHO MQTT介绍
项目开源地址：GitHub - eclipse/paho.mqtt.javascript: paho.mqtt.javascript
PAHO MQTT 客户端库支持多种平台，包括 Java、JavaScript、C、Python 等，使得开发者可以在不同的环境中使用相同的客户端代码来实现 MQTT 通信。
以下是 PAHO MQTT 客户端库的一些关键特性：
支持 MQTT 协议的各种版本：PAHO MQTT 客户端库提供了对 MQTT 3.1、3.1.1、5.0 版本的支持，使得开发者可以根据自己的需求选择合适的 MQTT 版本进行通信。
提供丰富的 API：PAHO MQTT 客户端库提供了丰富的 API，包括连接服务器、发布消息、订阅主题、处理消息等，开发者可以灵活地使用这些 API 来实现自己的 MQTT 应用程序。
支持 TLS/SSL 安全连接：PAHO MQTT 客户端库提供了对 TLS/SSL 安全连接的支持，使得 MQTT 通信可以在安全的网络环境下进行。
跨平台支持：PAHO MQTT 客户端库支持多种平台，包括 Java、JavaScript、C、Python 等，使得开发者可以在不同的环境中使用相同的客户端代码来实现 MQTT 通信。
持久化消息：PAHO MQTT 客户端库支持持久化消息，即使客户端断开连接，消息也可以被持久化，确保消息不会丢失。
原库的介绍：The Paho JavaScript Client is an MQTT browser-based client library written in Javascript that uses WebSockets to connect to an MQTT Broker.
原库的这个js库只支持WebSocket通信，本移植后续也增加支持tcp方式通信。
移植后的我的开源仓库地址：paho-mqtt: paho-mqtt移植
已发布到鸿蒙三方库市场，地址：OpenHarmony三方库中心仓
移植过程
移植参考第三方开源js库移植适配指南
从项目仓库paho.mqtt.javascript下载下来源码，打开paho-mqtt.js文件，首选看到的是以下代码：
(function ExportLibrary(root, factory) {
if (typeof exports === 'object' && typeof module === 'object') {
module.exports = factory();
} else if (typeof define === 'function' && define.amd) {
define(factory);
} else if (typeof exports === 'object') {
exports = factory();
} else {
if (typeof root.Paho === 'undefined') {
root.Paho = {};
}
root.Paho.MQTT = factory();
}
})( this,function LibraryFactory() {
// ......
});这段代码采用的是UMD (Universal Module Definition) 模块化规范，用于创建一个可以在CommonJS、AMD以及全局变量环境下都能正常加载和使用的JavaScript库。这种写法允许同一段代码在Node.js（CommonJS）、RequireJS（AMD）或非模块化的浏览器环境中运行。
代码结构解释如下：
ExportLibrary函数是UMD的实现部分，它接收两个参数：root和factory。
root通常指向全局对象，在浏览器中是window，在Node.js中是global。
factory是一个函数，当调用时会生成实际的库模块内容。
函数内部根据不同的环境条件执行相应的导出操作：
如果当前环境支持CommonJS（如Node.js），则将工厂函数的结果导出为module.exports。
如果当前环境支持AMD（如RequireJS），则使用define函数定义模块并传入工厂函数。
如果在非模块化环境且检测到exports存在（可能是在某些老版本的CommonJS兼容环境中），直接赋值给exports。
最后一种情况，如果没有上述任何模块系统，就将库挂载到全局对象上，这里是root.Paho.MQTT = factory();，即将factory函数返回的对象作为Paho命名空间下的MQTT属性暴露出去。
最后的括号( function LibraryFactory() { ... } )实际上是立即执行函数表达式（IIFE, Immediately Invoked Function Expression），在这里传递了LibraryFactory函数作为ExportLibrary函数的参数。但在这个示例中，LibraryFactory函数体为空，看起来可能是待填充实际库内容的部分。正常情况下，LibraryFactory应该包含了实际的库实现代码，以便在不同模块环境下正确初始化和导出。
在鸿蒙（HarmonyOS）的环境中，不存在名为global的全局对象。在浏览器环境中，通常全局对象是window，在Node.js 环境中则是global。然而，在鸿蒙操作系统中，全局对象的命名可能不同，或者可能不存在类似的全局对象用于在所有代码中共享变量和函数。因此在移植时，这里是需要大刀阔斧的砍掉和修改的。
接下来开始移植，引入鸿蒙的websocket库，在function LibraryFactory里，实例化websocket客户端实例。在 JavaScript 中，当多个模块 import 同一个模块时，导出的对象会被多次实例化吗？不会的，这是因为：
这是因为模块在 JavaScript 中是单例模式的，意味着无论import多少次，导出的对象都是同一个实例。下面是一个示例，假设有一个名为 MyClass 的类对象在 "myModule.js" 中被定义和导出：
// myModule.js
class MyClass {
constructor() {
// ...
}
// ...
}
export default new MyClass();
如果有多个模块import "myModule.js"，它们得到的是同一个 MyClass 的实例：
// moduleA.js
import myClassInstance from './myModule.js';
// 使用 myClassInstance
// moduleB.js
import myClassInstance from './myModule.js';
// 使用 myClassInstance
在 moduleA.js 和 moduleB.js 中 import 的 myClassInstance 实际上是同一个对象实例。
因此，不用担心多次 import 同一个模块会导致对象被多次实例化。
主要修改点：
ws.on('open', (err, value) => {
if(err){
this._trace("error:" + JSON.stringify(err));
}
this._trace("on open, status:" + JSON.stringify(value));
this._on_socket_open(value);
});
ws.on('message', (err, value) => {
this._trace("on message, message:" + value);
this._on_socket_message(value);
});
ws.on('close', (err, value) => {
this._trace("on close, code is " + value.code + ", reason is " + value.reason);
this._on_socket_close();
});
ws.on('error', (err) => {
this._trace("on error, error:" + JSON.stringify(err));
this._on_socket_error();
});
ws.connect(wsurl, (err, value) => {
if (!err) {
this._trace("Connected successfully");
} else {
this._trace("Connection failed. Err:" + JSON.stringify(err));
}
});
//ws.on('message',scope(this._on_socket_message, this));
//ws.on('close',scope(this._on_socket_close, this));
//ws.on('error',scope(this._on_socket_error, this));
整个代码量不大，很适合学习很适合学和参考。
移植后的完整代码,参见：https://blog.csdn.net/yyz_1987/article/details/136064930
const MQTT = LibraryFactory();
export default MQTT;
index.d.ts文件修改
index.d.ts文件如果不修改也是不能用的
export = MQTT;
declare global {
namespace MQTT {
/**
* The Quality of Service used to deliver a message.
* <dl>
* <dt>0 Best effort (default).</dt>
* <dt>1 At least once.</dt>
* <dt>2 Exactly once.</dt>
* </dl>
*/
type Qos = 0 | 1 | 2;
interface MQTTError {
/** A number indicating the nature of the error. */
errorCode: number;
/** Text describing the error */
errorMessage: string;
}
interface WithInvocationContext {
/**
* <code>invocationContext</code> as passed in with the corresponding field in the connectOptions or
* subscribeOptions.
*/
invocationContext: any;
}
interface ErrorWithInvocationContext extends MQTTError, WithInvocationContext {}
interface OnSubscribeSuccessParams extends WithInvocationContext {
grantedQos: Qos;
}
/**
* Called when the connect acknowledgement has been received from the server.
* @param o
* A single response object parameter is passed to the onSuccess callback containing the following fields:
* <li><code>invocationContext</code> as passed in with the corresponding field in the connectOptions.
*/
type OnSuccessCallback = (o: WithInvocationContext) => void;
/**
* Called when the subscribe acknowledgement has been received from the server.
* @param o
* A single response object parameter is passed to the onSuccess callback containing the following fields:
* <li><code>invocationContext</code> as passed in with the corresponding field in the connectOptions.
*/
type OnSubscribeSuccessCallback = (o: OnSubscribeSuccessParams) => void;
/**
* Called when the connect request has failed or timed out.
* @param e
* A single response object parameter is passed to the onFailure callback containing the following fields:
* <li><code>invocationContext</code> as passed in with the corresponding field in the connectOptions.
* <li><code>errorCode</code> a number indicating the nature of the error.
* <li><code>errorMessage</code> text describing the error.
*/
type OnFailureCallback = (e: ErrorWithInvocationContext) => void;
/**
* Called when a connection has been lost.
* @param error A single response object parameter is passed to the onConnectionLost callback containing the
* following fields:
* <li>errorCode
* <li>errorMessage
*/
type OnConnectionLostHandler = (error: MQTTError) => void;
/**
* Called when a message was delivered or has arrived.
* @param message The {@link Paho.MQTT.Message} that was delivered or has arrived.
*/
type OnMessageHandler = (message: Message) => void;
/**
* Attributes used with a connection.
*/
interface ConnectionOptions {
/**
* If the connect has not succeeded within this number of seconds, it is deemed to have failed.
* @default The default is 30 seconds.
*/
timeout?: number | undefined;
/** Authentication username for this connection. */
userName?: string | undefined;
/** Authentication password for this connection. */
password?: string | undefined;
/** Sent by the server when the client disconnects abnormally. */
willMessage?: Message | undefined;
/**
* The server disconnects this client if there is no activity for this number of seconds.
* @default The default value of 60 seconds is assumed if not set.
*/
keepAliveInterval?: number | undefined;
/**
* If true(default) the client and server persistent state is deleted on successful connect.
* @default true
*/
cleanSession?: boolean | undefined;
/** If present and true, use an SSL Websocket connection. */
useSSL?: boolean | undefined;
/** Passed to the onSuccess callback or onFailure callback. */
invocationContext?: any;
/**
* Called when the connect acknowledgement has been received from the server.
*/
onSuccess?: OnSuccessCallback | undefined;
/**
* Specifies the mqtt version to use when connecting
* <dl>
* <dt>3 - MQTT 3.1</dt>
* <dt>4 - MQTT 3.1.1 (default)</dt>
* </dl>
* @default 4
*/
mqttVersion?: 3 | 4 | undefined;
/**
* If set to true, will force the connection to use the selected MQTT Version or will fail to connect.
*/
mqttVersionExplicit?: boolean | undefined;
/**
* Called when the connect request has failed or timed out.
*/
onFailure?: OnFailureCallback | undefined;
/**
* If present this contains either a set of hostnames or fully qualified
* WebSocket URIs (ws://example.com:1883/mqtt), that are tried in order in place of the host and port
* paramater on the construtor. The hosts are tried one at at time in order until one of then succeeds.
*/
hosts?: string[] | undefined;
/**
* If present the set of ports matching the hosts. If hosts contains URIs, this property is not used.
*/
ports?: number[] | undefined;
/**
* Sets whether the client will automatically attempt to reconnect
* to the server if the connection is lost.
*/
reconnect?: boolean | undefined;
/**
* If present, should contain a list of fully qualified WebSocket
* uris (e.g. ws://mqtt.eclipseprojects.io:80/mqtt), that are tried
* in order in place of the host and port parameter of the
* construtor. The uris are tried one at a time in order until one
* of them succeeds. Do not use this in conjunction with hosts as
* the hosts array will be converted to uris and will overwrite this
* property.
*/
uris?: string[] | undefined;
}
/**
* Used to control a subscription
*/
interface SubscribeOptions {
/** the maximum qos of any publications sent as a result of making this subscription. */
qos?: Qos | undefined;
/** passed to the onSuccess callback or onFailure callback. */
invocationContext?: any;
/** called when the subscribe acknowledgement has been received from the server. */
onSuccess?: OnSubscribeSuccessCallback | undefined;
/** called when the subscribe request has failed or timed out. */
onFailure?: OnFailureCallback | undefined;
/**
* timeout which, if present, determines the number of seconds after which the onFailure calback is called.
* The presence of a timeout does not prevent the onSuccess callback from being called when the subscribe
* completes.
*/
timeout?: number | undefined;
}
interface UnsubscribeOptions {
/** passed to the onSuccess callback or onFailure callback. */
invocationContext?: any;
/** called when the unsubscribe acknowledgement has been received from the server. */
onSuccess?: OnSuccessCallback | undefined;
/** called when the unsubscribe request has failed or timed out. */
onFailure?: OnFailureCallback | undefined;
/**
* timeout which, if present, determines the number of seconds after which the onFailure calback is called.
* The presence of a timeout does not prevent the onSuccess callback from being called when the unsubscribe
* completes.
*/
timeout?: number | undefined;
}
interface TraceElement {
severity: "Debug";
message: string;
}
type TraceFunction = (element: TraceElement) => void;
/**
* The JavaScript application communicates to the server using a {@link Paho.MQTT.Client} object.
*
* Most applications will create just one Client object and then call its connect() method,
* however applications can create more than one Client object if they wish.
* In this case the combination of host, port and clientId attributes must be different for each Client object.
*
* The send, subscribe and unsubscribe methods are implemented as asynchronous JavaScript methods
* (even though the underlying protocol exchange might be synchronous in nature).
* This means they signal their completion by calling back to the application,
* via Success or Failure callback functions provided by the application on the method in question.
* Such callbacks are called at most once per method invocation and do not persist beyond the lifetime
* of the script that made the invocation.
*
* In contrast there are some callback functions, most notably {@link onMessageArrived},
* that are defined on the {@link Paho.MQTT.Client} object.
* These may get called multiple times, and aren't directly related to specific method invocations made by the
* client.
*/
class Client {
/** used when connecting to the server. */
readonly clientId: string;
/** the server's DNS hostname or dotted decimal IP address. */
readonly host: string;
/** the server's path. */
readonly path: string;
/** the server's port. */
readonly port: number;
/** function called with trace information, if set */
trace?: TraceFunction | undefined;
/**
* called when a connection has been lost. after a connect() method has succeeded.
* Establish the call back used when a connection has been lost. The connection may be
* lost because the client initiates a disconnect or because the server or network
* cause the client to be disconnected. The disconnect call back may be called without
* the connectionComplete call back being invoked if, for example the client fails to
* connect.
* A single response object parameter is passed to the onConnectionLost callback containing the following
* fields:
* <li>errorCode
* <li>errorMessage
*/
onConnectionLost: OnConnectionLostHandler;
/**
* called when a message has been delivered.
* All processing that this Client will ever do has been completed. So, for example,
* in the case of a Qos=2 message sent by this client, the PubComp flow has been received from the server
* and the message has been removed from persistent storage before this callback is invoked.
* Parameters passed to the onMessageDelivered callback are:
* <li>{@link Paho.MQTT.Message} that was delivered.
*/
onMessageDelivered: OnMessageHandler;
/**
* called when a message has arrived in this Paho.MQTT.client.
* Parameters passed to the onMessageArrived callback are:
* <li> {@link Paho.MQTT.Message} that has arrived.
*/
onMessageArrived: OnMessageHandler;
/**
* @param host - the address of the messaging server as a DNS name or dotted decimal IP address.
* @param port - the port number to connect to
* @param path - the path on the host to connect to - only used if host is not a URI. Default: '/mqtt'.
* @param clientId - the Messaging client identifier, between 1 and 23 characters in length.
*/
constructor(host: string, port: number, path: string, clientId: string); // tslint:disable-line unified-signatures (these cannot actually be neatly unified)
/**
* @param host - the address of the messaging server as a DNS name or dotted decimal IP address.
* @param port - the port number to connect to
* @param clientId - the Messaging client identifier, between 1 and 23 characters in length.
*/
constructor(host: string, port: number, clientId: string);
/**
* @param hostUri - the address of the messaging server as a fully qualified WebSocket URI
* @param clientId - the Messaging client identifier, between 1 and 23 characters in length.
*/
constructor(hostUri: string, clientId: string);
/**
* Connect this Messaging client to its server.
* @throws {InvalidState} if the client is not in disconnected state. The client must have received
* connectionLost or disconnected before calling connect for a second or subsequent time.
*/
connect(connectionOptions?: ConnectionOptions): void;
/**
* Normal disconnect of this Messaging client from its server.
*
* @throws {InvalidState} if the client is already disconnected.
*/
disconnect(): void;
/**
* @returns True if the client is currently connected
*/
isConnected(): boolean;
/**
* Get the contents of the trace log.
*
* @return tracebuffer containing the time ordered trace records.
*/
getTraceLog(): any[];
/**
* Start tracing.
*/
startTrace(): void;
/**
* Stop tracing.
*/
stopTrace(): void;
/**
* Send a message to the consumers of the destination in the Message.
*
* @param message - <b>mandatory</b> The {@link Paho.MQTT.Message} object to be sent.
* @throws {InvalidState} if the client is not connected.
*/
send(message: Message): void;
/**
* Send a message to the consumers of the destination in the Message.
*
* @param topic - <b>mandatory</b> The name of the destination to which the message is to be sent.
* @param payload - The message data to be sent.
* @param qos The Quality of Service used to deliver the message.
* <dl>
* <dt>0 Best effort (default).
* <dt>1 At least once.
* <dt>2 Exactly once.
* </dl>
* @param retained If true, the message is to be retained by the server and delivered to both
* current and future subscriptions. If false the server only delivers the message to current subscribers,
* this is the default for new Messages. A received message has the retained boolean set to true if the
* message was published with the retained boolean set to true and the subscrption was made after the
* message has been published.
* @throws {InvalidState} if the client is not connected.
*/
send(topic: string, payload: string | ArrayBuffer, qos?: Qos, retained?: boolean): void;
/**
* Subscribe for messages, request receipt of a copy of messages sent to the destinations described by the
* filter.
*
* @param filter A filter describing the destinations to receive messages from.
* @param subcribeOptions Used to control the subscription
* @throws {InvalidState} if the client is not in connected state.
*/
subscribe(filter: string, subcribeOptions?: SubscribeOptions): void;
/**
* Unsubscribe for messages, stop receiving messages sent to destinations described by the filter.
*
* @param filter - describing the destinations to receive messages from.
* @param unsubscribeOptions - used to control the subscription
* @throws {InvalidState} if the client is not in connected state.
*/
unsubscribe(filter: string, unsubcribeOptions?: UnsubscribeOptions): void;
}
type TypedArray =
| Int8Array
| Uint8Array
| Uint8ClampedArray
| Int16Array
| Uint16Array
| Int32Array
| Uint32Array
| Float32Array
| Float64Array;
/**
* An application message, sent or received.
*/
class Message {
/**
* The name of the destination to which the message is to be sent
* (for messages about to be sent) or the name of the destination from which the message has been received.
* (for messages received by the onMessage function).
*/
destinationName: string;
/**
* If true, this message might be a duplicate of one which has already been received.
* This is only set on messages received from the server.
*/
readonly duplicate: boolean;
/**
* The payload.
* @return if payload is a string. Return the original otherwise.
*/
readonly payloadBytes: ArrayBuffer | TypedArray;
/**
* The payload as a string if the payload consists of valid UTF-8 characters.
* @throws {Error} if the payload is not valid UTF-8
*/
readonly payloadString: string;
/**
* The Quality of Service used to deliver the message.
* <dl>
* <dt>0 Best effort (default).
* <dt>1 At least once.
* <dt>2 Exactly once.
* </dl>
*
* @default 0
*/
qos: Qos;
/**
* If true, the message is to be retained by the server and delivered to both current and future
* subscriptions. If false the server only delivers the message to current subscribers, this is the default
* for new Messages. A received message has the retained boolean set to true if the message was published
* with the retained boolean set to true and the subscription was made after the message has been published.
*
* @default false
*/
retained: boolean;
/**
* @param payload The message data to be sent.
*/
constructor(payload: string | ArrayBuffer | TypedArray);
}
}
}WebSocket在OpenHarmony上的实现过程
如下图：
原文作者：；如有侵权，请联系管理员删除。
最新文章
- 别再手动配对了！用鸿蒙分布式软总线RPC，5分钟搞定手机远程调用平板摄像头
- HarmonyOS分布式输入法开发：实现多设备无缝输入体验
- 【HarmonyOS 7开发者前瞻】09 HarmonyOS 6 项目升级 API 26 前，先检查这张兼容性清单
- 星闪NearLink vs 蓝牙5.3：实测对比谁才是智能家居的最佳无线方案？
- 鸿蒙设备描述元数据具体如何工作？
- 鸿蒙系统内核层的微内核架构
- 鸿蒙设备开发之OpenHarmony之分布式软总线json_payload.c(一)
- 6、构建安全容器镜像与应用支持服务的实践指南
- 一文彻底搞清楚HarmonyOS NEXT中的this
- 鸿蒙OS开发：【一次开发，多端部署】（响应式布局）
