anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化imclient代码

Browse Source
This commit is contained in:
heavyrain.lee
2019-05-15 23:02:55 +08:00
parent e955153d53
commit 4661b8c856
6 changed files with 44 additions and 607 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/cn/wildfirechat/client/IMClient.java
+40 -6
View File
@@ -34,6 +34,7 @@ public class IMClient implements Listener {
private final int port;
private ConnectionStatusCallback connectionStatusCallback;
private ReceiveMessageCallback receiveMessageCallback;
private ConnectionStatus connectionStatus;
protected String mqttServerIp;
protected long mqttServerPort;
@@ -102,9 +103,11 @@ public class IMClient implements Listener {
connection.listener(this);
//connecting
connectionStatus = ConnectionStatus_Connecting;
if(connectionStatusCallback != null) {
connectionStatusCallback.onConnectionStatusChanged(ConnectionStatus_Connecting);
connectionStatusCallback.onConnectionStatusChanged(connectionStatus);
}
connection.connect(new Callback<byte[]>() {
@Override
public void onSuccess(byte[] value) {
@@ -121,16 +124,18 @@ public class IMClient implements Listener {
System.out.println("on connect success");
//connected
connectionStatus = ConnectionStatus_Connected;
if(connectionStatusCallback != null) {
connectionStatusCallback.onConnectionStatusChanged(ConnectionStatus_Connected);
connectionStatusCallback.onConnectionStatusChanged(connectionStatus);
}
}
@Override
public void onFailure(Throwable value) {
System.out.println("on connect failure");
connectionStatus = ConnectionStatus_Unconnected;
if(connectionStatusCallback != null) {
connectionStatusCallback.onConnectionStatusChanged(ConnectionStatus_Unconnected);
connectionStatusCallback.onConnectionStatusChanged(connectionStatus);
}
}
});
@@ -142,6 +147,9 @@ public class IMClient implements Listener {
}
}
public void disconnect(boolean clearSession, final Callback<Void> onComplete) {
this.connection.disconnect(clearSession, onComplete);
}
public void sendMessage(WFCMessage.Conversation conversation, WFCMessage.MessageContent messageContent, final SendMessageCallback callback) {
WFCMessage.Message message = WFCMessage.Message.newBuilder().setConversation(conversation).setContent(messageContent).setFromUser(userId).build();
byte[] data = message.toByteArray();
@@ -249,16 +257,18 @@ public class IMClient implements Listener {
@Override
public void onConnected() {
System.out.println("onConnected");
connectionStatus = ConnectionStatus_Connecting;
if(connectionStatusCallback != null) {
connectionStatusCallback.onConnectionStatusChanged(ConnectionStatus_Connecting);
connectionStatusCallback.onConnectionStatusChanged(connectionStatus);
}
}
@Override
public void onDisconnected() {
System.out.println("onDisconnected");
connectionStatus = ConnectionStatus_Unconnected;
if(connectionStatusCallback != null) {
connectionStatusCallback.onConnectionStatusChanged(ConnectionStatus_Unconnected);
connectionStatusCallback.onConnectionStatusChanged(connectionStatus);
}
}
@@ -319,6 +329,30 @@ public class IMClient implements Listener {
this.receiveMessageCallback = receiveMessageCallback;
}
public String getUserId() {
return userId;
}
public String getToken() {
return token;
}
public String getClientId() {
return clientId;
}
public String getHost() {
return host;
}
public int getPort() {
return port;
}
public ConnectionStatus getConnectionStatus() {
return connectionStatus;
}
public static void main(String[] args) {
//token与userid和clientid是绑定的，使用时一定要传入正确的userid和clientid，不然会认为token非法
IMClient client = new IMClient("yzyOyOKK", "7SJk13q+YdHHe6EwDzry9BKogxTNf3UgtYj50cBTZgWNkNuxEkiqg2koKg0lXViONIX1LmwCR1jN0Mw8hvk6KGpiSKFi+IRaRkIb3mNzgIfrq4afhyIHaQfa2HOfsi6Ws+9YobkdDgdq7W70bEdVfiCSU9+JOIY449nxZzfg2Zw=", "DD72C212-26C7-4B38-A5FC-88550896B170", "192.168.1.101", 80);
@@ -338,7 +372,7 @@ public class IMClient implements Listener {
client.setConnectionStatusCallback((ConnectionStatus newStatus) -> {
if (newStatus == ConnectionStatus_Connected) {
WFCMessage.Conversation conversation = WFCMessage.Conversation.newBuilder().setType(0).setTarget("yzyOyOKK").setLine(0).build();
WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.newBuilder().setContent("helloworld").setType(1).build();
WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.newBuilder().setSearchableContent("helloworld").setType(1).build();
client.sendMessage(conversation, messageContent, new SendMessageCallback() {
@Override
public void onSuccess(long messageUid, long timestamp) {
client/src/main/java/org/fusesource/mqtt/cli/Listener.java
-272
View File
@@ -1,272 +0,0 @@
/**
* Copyright (C) 2010-2012, FuseSource Corp. All rights reserved.
*
* http://fusesource.com
*
* 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 org.fusesource.mqtt.cli;
import org.fusesource.mqtt.client.*;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.hawtdispatch.Task;
import org.fusesource.mqtt.client.*;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.concurrent.CountDownLatch;
/**
* <p>
* </p>
*
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
public class Listener {
private final MQTT mqtt = new MQTT();
private final ArrayList<Topic> topics = new ArrayList<Topic>();
private boolean debug;
private boolean showTopic;
private static void displayHelpAndExit(int exitCode) {
stdout("");
stdout("This is a simple mqtt client that will subscribe to topics and print all messages it receives.");
stdout("");
stdout("Arguments: [-h host] [-k keepalive] [-c] [-i id] [-u username [-p password]]");
stdout(" [--will-topic topic [--will-payload payload] [--will-qos qos] [--will-retain]]");
stdout(" [-d] [-s]");
stdout(" ( [-q qos] -t topic )+");
stdout("");
stdout("");
stdout(" -h : mqtt host uri to connect to. Defaults to tcp://localhost:1883.");
stdout(" -k : keep alive in seconds for this client. Defaults to 60.");
stdout(" -c : disable 'clean session' (store subscription and pending messages when client disconnects).");
stdout(" -i : id to use for this client. Defaults to a random id.");
stdout(" -u : provide a username (requires MQTT 3.1 broker)");
stdout(" -p : provide a password (requires MQTT 3.1 broker)");
stdout(" --will-topic : the topic on which to publish the client Will.");
stdout(" --will-payload : payload for the client Will, which is sent by the broker in case of");
stdout(" unexpected disconnection. If not given and will-topic is set, a zero");
stdout(" length message will be sent.");
stdout(" --will-qos : QoS level for the client Will.");
stdout(" --will-retain : if given, make the client Will retained.");
stdout(" -d : dispaly debug info on stderr");
stdout(" -s : show message topics in output");
stdout(" -q : quality of service level to use for the subscription. Defaults to 0.");
stdout(" -t : mqtt topic to subscribe to. May be repeated multiple times.");
stdout(" -v : MQTT version to use 3.1 or 3.1.1. (default: 3.1)");
stdout("");
System.exit(exitCode);
}
private static void stdout(Object x) {
System.out.println(x);
}
private static void stderr(Object x) {
System.err.println(x);
}
private static String shift(LinkedList<String> argl) {
if(argl.isEmpty()) {
stderr("Invalid usage: Missing argument");
displayHelpAndExit(1);
}
return argl.removeFirst();
}
public static void main(String[] args) throws Exception {
Listener main = new Listener();
// Process the arguments
QoS qos = QoS.AT_MOST_ONCE;
LinkedList<String> argl = new LinkedList<String>(Arrays.asList(args));
while (!argl.isEmpty()) {
try {
String arg = argl.removeFirst();
if ("--help".equals(arg)) {
displayHelpAndExit(0);
} else if ("-v".equals(arg)) {
main.mqtt.setVersion(shift(argl));
} else if ("-h".equals(arg)) {
main.mqtt.setHost(shift(argl));
} else if ("-k".equals(arg)) {
main.mqtt.setKeepAlive(Short.parseShort(shift(argl)));
} else if ("-c".equals(arg)) {
main.mqtt.setCleanSession(false);
} else if ("-i".equals(arg)) {
main.mqtt.setClientId(shift(argl));
} else if ("-u".equals(arg)) {
main.mqtt.setUserName(shift(argl));
} else if ("-p".equals(arg)) {
main.mqtt.setPassword(shift(argl));
} else if ("--will-topic".equals(arg)) {
main.mqtt.setWillTopic(shift(argl));
} else if ("--will-payload".equals(arg)) {
main.mqtt.setWillMessage(shift(argl));
} else if ("--will-qos".equals(arg)) {
int v = Integer.parseInt(shift(argl));
if( v > QoS.values().length ) {
stderr("Invalid qos value : " + v);
displayHelpAndExit(1);
}
main.mqtt.setWillQos(QoS.values()[v]);
} else if ("--will-retain".equals(arg)) {
main.mqtt.setWillRetain(true);
} else if ("-d".equals(arg)) {
main.debug = true;
} else if ("-s".equals(arg)) {
main.showTopic = true;
} else if ("-q".equals(arg)) {
int v = Integer.parseInt(shift(argl));
if( v > QoS.values().length ) {
stderr("Invalid qos value : " + v);
displayHelpAndExit(1);
}
qos = QoS.values()[v];
} else if ("-t".equals(arg)) {
main.topics.add(new Topic(shift(argl), qos));
} else {
stderr("Invalid usage: unknown option: " + arg);
displayHelpAndExit(1);
}
} catch (NumberFormatException e) {
stderr("Invalid usage: argument not a number");
displayHelpAndExit(1);
}
}
if (main.topics.isEmpty()) {
stderr("Invalid usage: no topics specified.");
displayHelpAndExit(1);
}
main.execute();
System.exit(0);
}
private void execute() {
final CallbackConnection connection = mqtt.callbackConnection();
final CountDownLatch done = new CountDownLatch(1);
// Handle a Ctrl-C event cleanly.
Runtime.getRuntime().addShutdownHook(new Thread(){
@Override
public void run() {
setName("MQTT client shutdown");
if(debug) {
stderr("Disconnecting the client.");
}
connection.getDispatchQueue().execute(new Task() {
public void run() {
connection.disconnect(new Callback<Void>() {
public void onSuccess(Void value) {
done.countDown();
}
public void onFailure(Throwable value) {
done.countDown();
}
});
}
});
}
});
connection.listener(new org.fusesource.mqtt.client.Listener() {
public void onConnected() {
if (debug) {
stderr("Connected");
}
}
public void onDisconnected() {
if (debug) {
stderr("Disconnected");
}
}
public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
try {
if (showTopic) {
stdout("");
stdout("Topic: " + topic);
body.writeTo(System.out);
stdout("");
} else {
body.writeTo(System.out);
}
ack.run();
} catch (IOException e) {
onFailure(e);
}
}
public void onFailure(Throwable value) {
if (debug) {
value.printStackTrace();
} else {
stderr(value);
}
System.exit(2);
}
});
connection.resume();
connection.connect(new Callback<byte[]>() {
public void onFailure(Throwable value) {
if (debug) {
value.printStackTrace();
} else {
stderr(value);
}
System.exit(2);
}
public void onSuccess(byte[] value) {
final Topic[] ta = topics.toArray(new Topic[topics.size()]);
connection.subscribe(ta, new Callback<byte[]>() {
public void onSuccess(byte[] value) {
if(debug) {
for (int i = 0; i < value.length; i++) {
stderr("Subscribed to Topic: " + ta[i].name() + " with QoS: " + QoS.values()[value[i]]);
}
}
}
public void onFailure(Throwable value) {
stderr("Subscribe failed: " + value);
if(debug) {
value.printStackTrace();
}
System.exit(2);
}
});
}
});
try {
done.await();
} catch (Exception e) {
e.printStackTrace();
}
System.exit(0);
}
}
client/src/main/java/org/fusesource/mqtt/cli/Publisher.java
-325
View File
@@ -1,325 +0,0 @@
/**
* Copyright (C) 2010-2012, FuseSource Corp. All rights reserved.
*
* http://fusesource.com
*
* 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 org.fusesource.mqtt.cli;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.Listener;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.hawtbuf.AsciiBuffer;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.ByteArrayOutputStream;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.hawtdispatch.Task;
import java.io.File;
import java.io.RandomAccessFile;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* <p>
* </p>
*
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
public class Publisher {
private final MQTT mqtt = new MQTT();
private QoS qos = QoS.AT_MOST_ONCE;
private UTF8Buffer topic;
private Buffer body;
private boolean debug;
private boolean retain;
private long count = 1;
private long sleep;
private boolean prefixCounter;
private static void displayHelpAndExit(int exitCode) {
stdout("");
stdout("This is a simple mqtt client that will publish to a topic.");
stdout("");
stdout("Arguments: [-h host] [-k keepalive] [-c] [-i id] [-u username [-p password]]");
stdout(" [--will-topic topic [--will-payload payload] [--will-qos qos] [--will-retain]]");
stdout(" [-d] [-n count] [-s sleep] [-q qos] [-r] -t topic ( -pc | -m message | -z | -f file )");
stdout("");
stdout("");
stdout(" -h : mqtt host uri to connect to. Defaults to tcp://localhost:1883.");
stdout(" -k : keep alive in seconds for this client. Defaults to 60.");
stdout(" -c : disable 'clean session'.");
stdout(" -i : id to use for this client. Defaults to a random id.");
stdout(" -u : provide a username (requires MQTT 3.1 broker)");
stdout(" -p : provide a password (requires MQTT 3.1 broker)");
stdout(" --will-topic : the topic on which to publish the client Will.");
stdout(" --will-payload : payload for the client Will, which is sent by the broker in case of");
stdout(" unexpected disconnection. If not given and will-topic is set, a zero");
stdout(" length message will be sent.");
stdout(" --will-qos : QoS level for the client Will.");
stdout(" --will-retain : if given, make the client Will retained.");
stdout(" -d : display debug info on stderr");
stdout(" -n : the number of times to publish the message");
stdout(" -s : the number of milliseconds to sleep between publish operations (defaut: 0)");
stdout(" -q : quality of service level to use for the publish. Defaults to 0.");
stdout(" -r : message should be retained.");
stdout(" -t : mqtt topic to publish to.");
stdout(" -m : message payload to send.");
stdout(" -z : send a null (zero length) message.");
stdout(" -f : send the contents of a file as the message.");
stdout(" -pc : prefix a message counter to the message");
stdout(" -v : MQTT version to use 3.1 or 3.1.1. (default: 3.1)");
stdout("");
System.exit(exitCode);
}
private static void stdout(Object x) {
System.out.println(x);
}
private static void stderr(Object x) {
System.err.println(x);
}
private static String shift(LinkedList<String> argl) {
if(argl.isEmpty()) {
stderr("Invalid usage: Missing argument");
displayHelpAndExit(1);
}
return argl.removeFirst();
}
public static void main(String[] args) throws Exception {
Publisher main = new Publisher();
// Process the arguments
LinkedList<String> argl = new LinkedList<String>(Arrays.asList(args));
while (!argl.isEmpty()) {
try {
String arg = argl.removeFirst();
if ("--help".equals(arg)) {
displayHelpAndExit(0);
} else if ("-v".equals(arg)) {
main.mqtt.setVersion(shift(argl));
} else if ("-h".equals(arg)) {
main.mqtt.setHost(shift(argl));
} else if ("-k".equals(arg)) {
main.mqtt.setKeepAlive(Short.parseShort(shift(argl)));
} else if ("-c".equals(arg)) {
main.mqtt.setCleanSession(false);
} else if ("-i".equals(arg)) {
main.mqtt.setClientId(shift(argl));
} else if ("-u".equals(arg)) {
main.mqtt.setUserName(shift(argl));
} else if ("-p".equals(arg)) {
main.mqtt.setPassword(shift(argl));
} else if ("--will-topic".equals(arg)) {
main.mqtt.setWillTopic(shift(argl));
} else if ("--will-payload".equals(arg)) {
main.mqtt.setWillMessage(shift(argl));
} else if ("--will-qos".equals(arg)) {
int v = Integer.parseInt(shift(argl));
if( v > QoS.values().length ) {
stderr("Invalid qos value : " + v);
displayHelpAndExit(1);
}
main.mqtt.setWillQos(QoS.values()[v]);
} else if ("--will-retain".equals(arg)) {
main.mqtt.setWillRetain(true);
} else if ("-d".equals(arg)) {
main.debug = true;
} else if ("-n".equals(arg)) {
main.count = Long.parseLong(shift(argl));
} else if ("-s".equals(arg)) {
main.sleep = Long.parseLong(shift(argl));
} else if ("-q".equals(arg)) {
int v = Integer.parseInt(shift(argl));
if( v > QoS.values().length ) {
stderr("Invalid qos value : " + v);
displayHelpAndExit(1);
}
main.qos = QoS.values()[v];
} else if ("-r".equals(arg)) {
main.retain = true;
} else if ("-t".equals(arg)) {
main.topic = new UTF8Buffer(shift(argl));
} else if ("-m".equals(arg)) {
main.body = new UTF8Buffer(shift(argl)+"\n");
} else if ("-z".equals(arg)) {
main.body = new UTF8Buffer("");
} else if ("-f".equals(arg)) {
File file = new File(shift(argl));
RandomAccessFile raf = new RandomAccessFile(file, "r");
try {
byte[] data = new byte[(int) raf.length()];
raf.seek(0);
raf.readFully(data);
main.body = new Buffer(data);
} finally {
raf.close();
}
} else if ("-pc".equals(arg)) {
main.prefixCounter = true;
} else {
stderr("Invalid usage: unknown option: " + arg);
displayHelpAndExit(1);
}
} catch (NumberFormatException e) {
stderr("Invalid usage: argument not a number");
displayHelpAndExit(1);
}
}
if (main.topic == null) {
stderr("Invalid usage: no topic specified.");
displayHelpAndExit(1);
}
if (main.body == null) {
stderr("Invalid usage: -z -m or -f must be specified.");
displayHelpAndExit(1);
}
main.execute();
System.exit(0);
}
private void execute() {
final CallbackConnection connection = mqtt.callbackConnection();
final CountDownLatch done = new CountDownLatch(1);
// Handle a Ctrl-C event cleanly.
Runtime.getRuntime().addShutdownHook(new Thread(){
@Override
public void run() {
setName("MQTT client shutdown");
connection.getDispatchQueue().execute(new Task() {
public void run() {
connection.disconnect(new Callback<Void>() {
public void onSuccess(Void value) {
done.countDown();
}
public void onFailure(Throwable value) {
done.countDown();
}
});
}
});
}
});
connection.listener(new Listener() {
public void onConnected() {
if (debug) {
stderr("Connected");
}
}
public void onDisconnected() {
if (debug) {
stderr("Disconnected");
}
}
public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
}
public void onFailure(Throwable value) {
if (debug) {
value.printStackTrace();
} else {
stderr(value);
}
System.exit(2);
}
});
connection.resume();
connection.connect(new Callback<byte[]>() {
public void onFailure(Throwable value) {
if (debug) {
value.printStackTrace();
} else {
stderr(value);
}
System.exit(2);
}
public void onSuccess(byte[] value) {
}
});
new Task() {
private long sent = 0;
public void run() {
final Task publish = this;
Buffer message = body;
if(prefixCounter) {
long id = sent + 1;
ByteArrayOutputStream os = new ByteArrayOutputStream(message.length + 15);
os.write(new AsciiBuffer(Long.toString(id)));
os.write(':');
os.write(body);
message = os.toBuffer();
}
connection.publish(topic, message, qos, retain, new Callback<byte[]>() {
public void onSuccess(byte[] value) {
sent ++;
if(debug) {
stdout("Sent message #"+sent);
}
if( sent < count ) {
if(sleep>0) {
System.out.println("Sleeping");
connection.getDispatchQueue().executeAfter(sleep, TimeUnit.MILLISECONDS, publish);
} else {
connection.getDispatchQueue().execute(publish);
}
} else {
connection.disconnect(new Callback<Void>() {
public void onSuccess(Void value) {
done.countDown();
}
public void onFailure(Throwable value) {
done.countDown();
}
});
}
}
public void onFailure(Throwable value) {
stderr("Publish failed: " + value);
if(debug) {
value.printStackTrace();
}
System.exit(2);
}
});
}
}.run();
try {
done.await();
} catch (Exception e) {
e.printStackTrace();
}
System.exit(0);
}
}
client/src/main/java/org/fusesource/mqtt/client/CallbackConnection.java
+2 -2
View File
@@ -554,7 +554,7 @@ public class CallbackConnection {
return failure;
}
public void disconnect(final Callback<Void> onComplete) {
public void disconnect(boolean clearSession, final Callback<Void> onComplete) {
if( disconnected ) {
if(onComplete!=null){
onComplete.onSuccess(null);
@@ -611,7 +611,7 @@ public class CallbackConnection {
// Pop the frame into a request so it we get notified
// of any failures so we continue to stop the transport.
if(transport!=null) {
MQTTFrame frame = new DISCONNECT().encode();
MQTTFrame frame = new DISCONNECT().encode().dup(clearSession);
send(new Request(getNextMessageId(), frame, cb));
} else {
cb.onSuccess(null);
client/src/main/java/org/fusesource/mqtt/client/FutureConnection.java
+1 -1
View File
@@ -151,7 +151,7 @@ public class FutureConnection {
final Promise<Void> future = new Promise<Void>();
next.getDispatchQueue().execute(new Task() {
public void run() {
next.disconnect(future);
next.disconnect(false, future);
}
});
return future;
distribution/src/main/resources/README.txt
+1 -1
View File
@@ -1 +1 @@
To run the Wildfire Chat distribution, untar the archive and run the script bin/wildfirechat.sh
部署方法请参考野火IM开发文档 http://docs.wildfirechat.cn