Repository navigation
Expand file tree
/
Copy pathStomp.java
More file actions
373 lines (335 loc) · 11 KB
/
Copy pathStomp.java
File metadata and controls
373 lines (335 loc) · 11 KB
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
package com.ejlchina.stomp;
import com.ejlchina.okhttps.OnCallback;
import com.ejlchina.okhttps.Platform;
import com.ejlchina.okhttps.WebSocket;
import com.ejlchina.okhttps.internal.WebSocketTask;
import okio.ByteString;
import java.util.*;
/**
* 基于 OkHttps websockt 的 Stomp 客户端
*/
public class Stomp {
private static final String TOPIC = "/topic";
private static final String QUEUE = "/queue";
public static final String SUPPORTED_VERSIONS = "1.1,1.2";
public static final String AUTO_ACK = "auto";
public static final String CLIENT_ACK = "client";
private final boolean autoAck;
private boolean connected;
private final WebSocketTask task;
private WebSocket websocket;
private boolean legacyWhitespace = false;
private final List<Subscriber> subscribers;
private OnCallback<Stomp> onConnected;
private OnCallback<WebSocket.Close> onDisconnected;
private OnCallback<Message> onError;
private final String disReceipt;
private Stomp(WebSocketTask task, boolean autoAck) {
this.task = task;
this.autoAck = autoAck;
this.subscribers = Collections.synchronizedList(new ArrayList<>());
this.disReceipt = UUID.randomUUID().toString();
}
/**
* 构建 Stomp 客户端（自动确定消息）
* @param task 底层的 WebSocket 连接
* @return Stomp
*/
public static Stomp over(WebSocketTask task) {
return over(task, true);
}
/**
* 构建 Stomp 客户端
* @param task 底层的 WebSocket 连接
* @param autoAck 是否自动确定消息
* @return Stomp
*/
public static Stomp over(WebSocketTask task, boolean autoAck) {
return new Stomp(task, autoAck);
}
/**
* @since 2.5.0
* @return 是否自动确认消息
*/
public boolean isAutoAck() {
return autoAck;
}
/**
* 连接 Stomp 服务器
* @return Stomp
*/
public Stomp connect() {
return connect(null);
}
/**
* 连接 Stomp 服务器
* @param headers Stomp 头信息
* @return Stomp
*/
public Stomp connect(List<Header> headers) {
if (connected) {
return this;
}
websocket = task.setOnOpen((ws, res) -> {
int pingSecs = task.pingSeconds();
int pongSecs = task.pongSeconds();
List<Header> cHeaders = new ArrayList<>();
cHeaders.add(new Header(Header.VERSION, SUPPORTED_VERSIONS));
if (pingSecs > 0 && pongSecs > 0) {
cHeaders.add(new Header(Header.HEART_BEAT, pingSecs * 1000 + "," + pongSecs * 1000));
}
if (headers != null) {
cHeaders.addAll(headers);
}
send(new Message(Commands.CONNECT, cHeaders, null));
})
.setOnMessage((ws, msg) -> {
Message message = Message.from(msg.toString());
if (message != null) {
receive(message);
}
})
.setOnClosed((ws, close) -> {
for (Subscriber subscriber : subscribers) {
subscriber.resetStatus();
}
if (onDisconnected != null) {
onDisconnected.on(close);
}
connected = false;
})
.listen();
return this;
}
/**
* @since 2.5.0
* @return 是否已连接
*/
public boolean isConnected() {
return connected && websocket.status() == WebSocket.STATUS_CONNECTED;
}
/**
* 断开连接
*/
public void disconnect() {
send(new Message(Commands.DISCONNECT, Collections.singletonList(new Header(Header.RECEIPT, disReceipt))));
}
/**
* 连接成功回调
* @param onConnected 连接成功回调
* @return Stomp
*/
public Stomp setOnConnected(OnCallback<Stomp> onConnected) {
this.onConnected = onConnected;
return this;
}
/**
* 连接断开回调
* @param onDisconnected 断开连接回调
* @return Stomp
*/
public Stomp setOnDisconnected(OnCallback<WebSocket.Close> onDisconnected) {
this.onDisconnected = onDisconnected;
return this;
}
/**
* 错误回调（服务器返回的错误信息）
* @param onError 错误回调
* @return Stomp
*/
public Stomp setOnError(OnCallback<Message> onError) {
this.onError = onError;
return this;
}
/**
* @since 2.5.0
* 发送消息到主题
* @param destination 目的地
* @param data 消息
*/
public void sendToTopic(String destination, String data) {
sendTo(TOPIC + destination, data);
}
/**
* @since 2.5.0
* 发送消息到队列
* @param destination 目的地
* @param data 消息
*/
public void sendToQueue(String destination, String data) {
sendTo(QUEUE + destination, data);
}
/**
* 发送消息到指定目的地
* @param destination 目的地
* @param data 消息
*/
public void sendTo(String destination, String data) {
send(new Message(Commands.SEND, Collections.singletonList(new Header(Header.DESTINATION, destination)), data));
}
/**
* 发送消息给服务器
* @param message 消息
*/
public void send(Message message) {
if (websocket == null) {
throw new IllegalArgumentException("You must call connect before send");
}
websocket.send(message.compile(legacyWhitespace));
}
/**
* 监听主题消息
* @param destination 监听地址
* @param callback 消息回调
* @return Stomp
*/
public Stomp topic(String destination, OnCallback<Message> callback) {
return topic(destination, null, callback);
}
/**
* 监听主题消息
* @param destination 监听地址
* @param headers 附加头信息
* @param callback 消息回调
* @return Stomp
*/
public Stomp topic(String destination, List<Header> headers, OnCallback<Message> callback) {
return subscribe(TOPIC + destination, headers, callback);
}
/**
* 监听队列消息
* @param destination 监听地址
* @param callback 消息回调
* @return Stomp
*/
public Stomp queue(String destination, OnCallback<Message> callback) {
return queue(destination, null, callback);
}
/**
* 监听队列消息
* @param destination 监听地址
* @param headers 附加头信息
* @param callback 消息回调
* @return Stomp
*/
public Stomp queue(String destination, List<Header> headers, OnCallback<Message> callback) {
return subscribe(QUEUE + destination, headers, callback);
}
/**
* 订阅消息
* @param destination 订阅地址
* @param headers 附加头信息
* @param callback 消息回调
* @return Stomp
*/
public synchronized Stomp subscribe(String destination, List<Header> headers, OnCallback<Message> callback) {
if (destination == null || destination.isEmpty()) {
throw new IllegalArgumentException("destination can not be empty!");
}
for (Subscriber s: subscribers) {
if (s.destinationEqual(destination)) {
throw new IllegalStateException("The destination [" + destination + "] has already been subscribed!");
}
}
Subscriber subscriber = new Subscriber(this, destination, callback, headers);
subscribers.add(subscriber);
subscriber.subscribe();
return this;
}
/**
* 确认收到某条消息
* @param message 服务器发过来的消息
*/
public void ack(Message message) {
Header subscription = message.header(Header.SUBSCRIPTION);
Header msgId = message.header(Header.MESSAGE_ID);
if (subscription != null || msgId != null) {
List<Header> headers = new ArrayList<>();
headers.add(subscription);
headers.add(msgId);
send(new Message(Commands.ACK, headers, null));
} else {
Platform.logError("subscription and message-id not found in " + message.toString() + ", so it can not be ack!");
}
}
/**
* 取消主题监听
* @param destination 监听地址
*/
public void untopic(String destination) {
unsubscribe(TOPIC + destination);
}
/**
* 取消队列监听
* @param destination 监听地址
*/
public void unqueue(String destination) {
unsubscribe(QUEUE + destination);
}
/**
* 取消订阅
* @param destination 订阅地址
*/
public synchronized void unsubscribe(String destination) {
Iterator<Subscriber> it = subscribers.iterator();
while (it.hasNext()) {
Subscriber s = it.next();
if (s.destinationEqual(destination)) {
s.unsubscribe();
it.remove();
break;
}
}
}
private void receive(Message msg) {
String command = msg.getCommand();
if (Commands.CONNECTED.equals(command)) {
String hbHeader = msg.headerValue(Header.HEART_BEAT);
int pingSecs = task.pingSeconds();
int pongSecs = task.pongSeconds();
if (hbHeader != null && (pingSecs > 0 || pongSecs > 0)) {
String[] heartbeats = hbHeader.split(",");
int pingSeconds = Integer.parseInt(heartbeats[1]) / 1000;
int pongSeconds = Integer.parseInt(heartbeats[0]) / 1000;
if (pingSeconds > 0 || pongSeconds > 0) {
if (task.pingSupplier() == null) {
task.pingSupplier(() -> ByteString.of((byte) 0x0A));
}
task.heatbeat(Math.max(pingSeconds, pingSecs), Math.max(pongSeconds, pongSecs));
}
}
synchronized (this) {
connected = true;
for (Subscriber s: subscribers) {
s.subscribe();
}
}
if (onConnected != null) {
onConnected.on(this);
}
} else if (Commands.MESSAGE.equals(command)) {
String id = msg.headerValue(Header.SUBSCRIPTION);
if (id != null) {
for (Subscriber s: subscribers) {
if (s.tryCallback(id, msg)) {
break;
}
}
}
} else if (Commands.RECEIPT.equals(command)) {
if (disReceipt.equals(msg.headerValue(Header.RECEIPT_ID))) {
// 断开连接
if (websocket != null) {
websocket.close(1000, "disconnect by user");
}
}
} else if (Commands.ERROR.equals(command)) {
if (onError != null) {
onError.on(msg);
}
}
}
public void setLegacyWhitespace(boolean legacyWhitespace) {
this.legacyWhitespace = legacyWhitespace;
}
}