1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
| import EventEmitter from 'eventemitter3'
| import {
| OPTIONS,
| } from './config'
| import {
| connect,
| MqttClient,
| IClientPublishOptions,
| IPublishPacket,
| Packet,
| ISubscriptionGrant,
| IClientOptions,
| } from 'mqtt/dist/mqtt.min'
|
| export class UranusMqtt extends EventEmitter {
| _url: string
| _options?: IClientOptions
| _client: MqttClient | null
| _hasInit: boolean
|
| constructor (url?: string, options?: IClientOptions) {
| super()
| this._url = url || ''
| this._options = options
| this._client = null
| this._hasInit = false
| }
|
| initMqtt = () => {
| // 仅初始化一次
| if (this._hasInit) return
| // 建立连接
| this._client = connect(this._url, {
| ...OPTIONS,
| ...this._options,
| })
| this._hasInit = true
| if (this._client) {
| this._client.on('reconnect', this._onReconnect)
| this._client.on('connect', () => {
| console.log('connect')
| })
| // 消息监听
| this._client.on('message', this._onMessage)
|
| // 连接关闭
| this._client.on('close', this._onClose)
|
| // 连接异常
| this._client.on('error', this._onError)
| }
| }
|
| // 发布
| publishMqtt = (topic: string, body: string | Buffer, opts?: IClientPublishOptions) => {
| if (!this._client?.connected) {
| this.initMqtt()
| }
| this._client?.publish(topic, body, opts || {}, (error?: Error, packet?: Packet) => {
| if (error) {
| window.console.error('mqtt publish error,', error, packet)
| }
| })
| }
|
| // 订阅
| subscribeMqtt = (topic: string) => {
| if (!this._client?.connected) {
| this.initMqtt()
| }
| window.console.log('subscribeMqtt>>>>>', topic)
| this._client?.subscribe(topic, (error: Error, granted: ISubscriptionGrant[]) => {
| window.console.log('mqtt subscribe,', error, granted)
| })
| }
|
| // 取消订阅
| unsubscribeMqtt = (topic: string) => {
| window.console.log('mqtt unsubscribeMqtt,', topic)
| this._client?.unsubscribe(topic)
| }
|
| // 关闭 mqtt 客户端
| destroyed = () => {
| window.console.log('mqtt destroyed')
| this._client?.end()
| }
|
| _onReconnect = () => {
| if (this._client) { window.console.error('mqtt reconnect,') }
| }
|
| _onMessage = (topic: string, payload: Buffer, packet: IPublishPacket) => {
| this.emit('onMessageMqtt', { topic, payload, packet })
| }
|
| _onClose = () => {
| // 连接异常关闭会自动重连
| window.console.error('mqtt close,')
| this.emit('onStatus', {
| status: 'close',
| })
| }
|
| _onError = (error: Error) => {
| // 连接错误会自动重连
| window.console.error('mqtt error,', error)
| this.emit('onStatus', {
| status: 'error',
| data: error,
| })
| }
| }
|
| export {
| IClientOptions,
| IPublishPacket,
| IClientPublishOptions,
| }
|
|
