From cfc97e602e453e4457d79d16d5be358a07010e7f Mon Sep 17 00:00:00 2001 From: "heavyrain.lee" Date: Wed, 15 May 2019 23:03:18 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84jmeter=E6=8F=92=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../net/xmeter/samplers/ClientManager.java | 32 +++ .../xmeter/samplers/ConnectionCallback.java | 42 ---- .../xmeter/samplers/ConnectionsManager.java | 72 ------- .../xmeter/samplers/SendMessageCallback.java | 54 ----- .../xmeter/samplers/SendMessageSampler.java | 190 ++++++++---------- 5 files changed, 117 insertions(+), 273 deletions(-) create mode 100755 wildfire_jmeter/src/main/java/net/xmeter/samplers/ClientManager.java delete mode 100755 wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionCallback.java delete mode 100755 wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionsManager.java delete mode 100755 wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageCallback.java diff --git a/wildfire_jmeter/src/main/java/net/xmeter/samplers/ClientManager.java b/wildfire_jmeter/src/main/java/net/xmeter/samplers/ClientManager.java new file mode 100755 index 00000000..482a02d7 --- /dev/null +++ b/wildfire_jmeter/src/main/java/net/xmeter/samplers/ClientManager.java @@ -0,0 +1,32 @@ +package net.xmeter.samplers; + +import java.util.concurrent.ConcurrentHashMap; + +import cn.wildfirechat.client.IMClient; + +public class ClientManager { + private ConcurrentHashMap clients = new ConcurrentHashMap<>(); + + + private static ClientManager clientManager = new ClientManager(); + private ClientManager() { + + } + + public static synchronized ClientManager getInstance() { + return clientManager; + } + + public void putClient(String key, IMClient client) { + clients.put(key, client); + } + + public IMClient getClient(String key) { + return this.clients.get(key); + } + + + public void removeClient(String key) { + this.clients.remove(key); + } +} diff --git a/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionCallback.java b/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionCallback.java deleted file mode 100755 index 4bb5777d..00000000 --- a/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionCallback.java +++ /dev/null @@ -1,42 +0,0 @@ -package net.xmeter.samplers; - -import java.text.MessageFormat; - -import org.apache.jorphan.logging.LoggingManager; -import org.apache.log.Logger; -import org.apache.log.Priority; -import org.fusesource.mqtt.client.Callback; -import org.fusesource.mqtt.client.CallbackConnection; - -public class ConnectionCallback implements Callback{ - private static Logger logger = LoggingManager.getLoggerForClass(); - private Object connLock; - private CallbackConnection connection; - private boolean connectionSucc = false; - - public ConnectionCallback(CallbackConnection connection, Object connLock) { - this.connection = connection; - this.connLock = connLock; - } - @Override - public void onSuccess(byte[] value) { - synchronized (connLock) { - logger.info(MessageFormat.format("The connection {0} is established successfully.", this.connection + Thread.currentThread().getName())); - connectionSucc = true; - this.connLock.notify(); - } - } - - @Override - public void onFailure(Throwable value) { - synchronized (connLock) { - connectionSucc = false; - logger.log(Priority.ERROR, value.getMessage(), value); - this.connLock.notify(); - } - } - - public boolean isConnectionSucc() { - return connectionSucc; - } -} diff --git a/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionsManager.java b/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionsManager.java deleted file mode 100755 index 153f20c4..00000000 --- a/wildfire_jmeter/src/main/java/net/xmeter/samplers/ConnectionsManager.java +++ /dev/null @@ -1,72 +0,0 @@ -package net.xmeter.samplers; - -import java.util.concurrent.ConcurrentHashMap; - -import org.fusesource.mqtt.client.CallbackConnection; -import org.fusesource.mqtt.client.MQTT; - -public class ConnectionsManager { - public static class ConnectionInfo { - public String userName; - public String token; - public String privateSecrect; - public String serverAddress; - public long serverPort; - - public ConnectionInfo(String userName, String token, String privateSecrect, String serverAddress, long serverPort) { - this.userName = userName; - this.token = token; - this.privateSecrect = privateSecrect; - this.serverAddress = serverAddress; - this.serverPort = serverPort; - } - } - - private ConcurrentHashMap connections = new ConcurrentHashMap<>(); - private ConcurrentHashMap connectionsStatus = new ConcurrentHashMap<>(); - private ConcurrentHashMap connectionInfos = new ConcurrentHashMap<>(); - - private static ConnectionsManager connectionsManager = new ConnectionsManager(); - private ConnectionsManager() { - - } - - public static synchronized ConnectionsManager getInstance() { - return connectionsManager; - } - - public CallbackConnection createConnection(String key, MQTT mqtt, ConnectionInfo connectionInfo) { - CallbackConnection conn = mqtt.callbackConnection(); - connections.put(key, conn); - connectionInfos.put(key, connectionInfo); - return conn; - } - - public CallbackConnection getConnection(String key) { - return this.connections.get(key); - } - - public ConnectionInfo getConnectionInfo(String key) { - return this.connectionInfos.get(key); - } - - public boolean containsConnection(String key) { - return connections.containsKey(key); - } - - public void removeConnection(String key) { - this.connections.remove(key); - this.connectionInfos.remove(key); - } - - public void setConnectionStatus(String key, Boolean status) { - connectionsStatus.put(key, status); - } - - public boolean getConnectionStatus(String key) { - if(!connectionsStatus.containsKey(key)) { - throw new RuntimeException("Cannot find conn status for key: " + key); - } - return connectionsStatus.get(key); - } -} diff --git a/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageCallback.java b/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageCallback.java deleted file mode 100755 index de310a11..00000000 --- a/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageCallback.java +++ /dev/null @@ -1,54 +0,0 @@ -package net.xmeter.samplers; - -import org.apache.jorphan.logging.LoggingManager; -import org.apache.log.Logger; -import org.apache.log.Priority; -import org.fusesource.mqtt.client.Callback; - -public class SendMessageCallback implements Callback{ - private static Logger logger = LoggingManager.getLoggerForClass(); - private boolean successful = false; - private Object connLock; - private int errorCode; - - public SendMessageCallback(Object connLock) { - this.connLock = connLock; - } - - @Override - public void onSuccess(byte[] value) { - //If QoS == 0, then the current thread is the same thread of caller thread. - //Else if QoS == 1 | 2, then the current thread is hawtdispatch-DEFAULT-x - if (value != null && value.length > 0) { - errorCode = value[0]&0xff; - } else { - onFailure(new Throwable("Send message with invalid data back")); - return; - } - if (errorCode != 0) { - onFailure(new Throwable("Send message failure with error code(" + errorCode + ")")); - return; - } - synchronized (connLock) { - this.successful = true; - connLock.notify(); - } - } - - @Override - public void onFailure(Throwable value) { - synchronized (connLock) { - this.successful = false; - logger.log(Priority.ERROR, value.getMessage(), value); - connLock.notify(); - } - } - - public boolean isSuccessful() { - return successful; - } - - public int getErrorCode() { - return errorCode; - } -} diff --git a/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageSampler.java b/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageSampler.java index c604330d..077910ff 100755 --- a/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageSampler.java +++ b/wildfire_jmeter/src/main/java/net/xmeter/samplers/SendMessageSampler.java @@ -1,12 +1,15 @@ package net.xmeter.samplers; import java.text.MessageFormat; +import java.util.ArrayList; +import java.util.List; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import javax.xml.bind.DatatypeConverter; +import cn.wildfirechat.client.IMClient; import cn.wildfirechat.proto.ProtoConstants; import cn.wildfirechat.proto.WFCMessage; import io.moquette.spi.impl.security.AES; @@ -22,10 +25,12 @@ import org.fusesource.hawtbuf.UTF8Buffer; import net.xmeter.Util; import org.fusesource.mqtt.client.Callback; -import org.fusesource.mqtt.client.CallbackConnection; import org.fusesource.mqtt.client.Listener; import org.fusesource.mqtt.client.QoS; +import static cn.wildfirechat.client.IMClient.ConnectionStatus.ConnectionStatus_Connected; +import static cn.wildfirechat.client.IMClient.ConnectionStatus.ConnectionStatus_Connecting; + public class SendMessageSampler extends AbstractMQTTSampler implements ThreadListener { /** @@ -34,11 +39,11 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis private static final long serialVersionUID = -4312341622759500786L; private transient static Logger logger = LoggingManager.getLoggerForClass(); private transient MQTT mqtt = new MQTT(); - private transient CallbackConnection connection = null; + private transient IMClient imClient = null; private String payload = null; private QoS qos_enum = QoS.AT_MOST_ONCE; private String topicName = ""; - private String connKey = ""; + private String clientKey = ""; public String getQOS() { return getPropertyAsString(QOS_LEVEL, String.valueOf(QOS_0)); @@ -127,76 +132,42 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis @Override public SampleResult sample(Entry arg0) { - this.connKey = getKey(); - if(connection == null) { - connection = ConnectionsManager.getInstance().getConnection(connKey); - if(connection != null) { - logger.info("Use the shared connection: " + connection); - ConnectionsManager.ConnectionInfo info = ConnectionsManager.getInstance().getConnectionInfo(connKey); - setUserName(info.userName); - token = info.token; - privateSecret = info.privateSecrect; - mqttServerIp = info.serverAddress; - mqttServerPort = info.serverPort; - } else { - try { - if (!getToken(getUserNameAuth())) { - throw new Exception("get token failure!!!"); + this.clientKey = getKey(); + + imClient = ClientManager.getInstance().getClient(clientKey); + if(imClient == null) { + try { + if (!getToken(getUserNameAuth(), getClientId())) { + throw new Exception("get token failure!!!"); + } + + imClient = new IMClient(getUserNameAuth(), token, getClientId(), getServer(), Integer.parseInt(getPort())); + + final Object lock = new Object(); + + + final List ret = new ArrayList<>(); + imClient.setConnectionStatusCallback(new IMClient.ConnectionStatusCallback() { + @Override + public void onConnectionStatusChanged(IMClient.ConnectionStatus newStatus) { + if (newStatus == ConnectionStatus_Connected) { + synchronized (lock) { + ret.add(true); + lock.notify(); + } + } } + }); - if (!route(getUserNameAuth(), token)) { - throw new Exception("route failure!!!"); - } + synchronized (lock) { + imClient.connect(); + lock.wait(Integer.parseInt(getConnKeepTime()) * 1000); + } - mqtt.setHost("tcp://" + mqttServerIp + ":" + mqttServerPort); - mqtt.setVersion(getMqttVersion()); - mqtt.setKeepAlive((short) Integer.parseInt(getConnKeepAlive())); - - mqtt.setClientId(getClientId()); - mqtt.setConnectAttemptsMax(Integer.parseInt(getConnAttamptMax())); - mqtt.setReconnectAttemptsMax(Integer.parseInt(getConnReconnAttamptMax())); - - mqtt.setUserName(getUserNameAuth()); - - byte[] password = AES.AESEncrypt(token, privateSecret); - mqtt.setPassword(new UTF8Buffer(password)); - - Object connLock = new Object(); - connection = ConnectionsManager.getInstance().createConnection(connKey, mqtt, new ConnectionsManager.ConnectionInfo(getUserNameAuth(), token, privateSecret, mqttServerIp, mqttServerPort)); - - connection.listener(new Listener() { - @Override - public void onConnected() { - - } - - @Override - public void onDisconnected() { - - } - - @Override - public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) { - logger.log(Priority.DEBUG, "receive publish with topic:" + topic); - ack.run(); - } - - @Override - public void onFailure(Throwable value) { - - } - }); - synchronized (connLock) { - ConnectionCallback callback = new ConnectionCallback(connection, connLock); - connection.connect(callback); - connLock.wait(TimeUnit.SECONDS.toMillis(Integer.parseInt(getConnTimeout()))); - ConnectionsManager.getInstance().setConnectionStatus(connKey, callback.isConnectionSucc()); - } - } catch (Exception e) { - logger.log(Priority.ERROR, e.getMessage(), e); - ConnectionsManager.getInstance().setConnectionStatus(connKey, false); - } - } + ClientManager.getInstance().putClient(clientKey, imClient); + } catch (Exception e) { + logger.log(Priority.ERROR, e.getMessage(), e); + } } payload = getMessage(); @@ -204,55 +175,66 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis payload = Util.generatePayload(120); } - SampleResult result = new SampleResult(); + final SampleResult result = new SampleResult(); result.setSampleLabel(getName()); - if(!ConnectionsManager.getInstance().getConnectionStatus(connKey)) { + if(ClientManager.getInstance().getClient(clientKey) == null) { result.sampleStart(); result.setSuccessful(false); result.sampleEnd(); - result.setResponseMessage(MessageFormat.format("Publish failed for connection {0}.", connection)); - result.setResponseData("Publish failed becasue the connection has not been established.".getBytes()); + result.setResponseMessage(MessageFormat.format("Publish failed for imClient {0}.", imClient)); + result.setResponseData("Publish failed becasue the imClient has not been established.".getBytes()); result.setResponseCode("500"); return result; } try { - byte[] toSend; result.sampleStart(); - final Object connLock = new Object(); - SendMessageCallback pubCallback = new SendMessageCallback(connLock); + final Object lock = new Object(); - WFCMessage.Message message = WFCMessage.Message.newBuilder() - .setConversation(WFCMessage.Conversation.newBuilder().setType(getConvType().equals(CONV_TYPE_SINGLE) ? ProtoConstants.ConversationType.ConversationType_Private : ProtoConstants.ConversationType.ConversationType_Group).setTarget(getTarget()).setLine(0).build()) - .setFromUser("") - .setContent(WFCMessage.MessageContent.newBuilder().setType(1).setSearchableContent(getMessage()).build()) - .build(); - toSend = message.toByteArray(); - toSend = AES.AESEncrypt(toSend, privateSecret); + WFCMessage.Conversation conversation = WFCMessage.Conversation.newBuilder().setType(0).setTarget("yzyOyOKK").setLine(0).build(); + WFCMessage.MessageContent messageContent = WFCMessage.MessageContent.newBuilder().setType(1).setSearchableContent(getMessage()).build(); - //wildfire send message use Qos1 and MS topic - topicName = "MS"; - synchronized (connLock) { - connection.publish(topicName, toSend, QoS.AT_LEAST_ONCE, false, pubCallback); - connLock.wait(); + final List ret = new ArrayList<>(); + final List error = new ArrayList<>(); + synchronized (lock) { + imClient.sendMessage(conversation, messageContent, new IMClient.SendMessageCallback() { + @Override + public void onSuccess(long messageUid, long timestamp) { + System.out.println("send success"); + ret.add(messageUid); + synchronized (lock) { + lock.notify(); + } + } + + @Override + public void onFailure(int errorCode) { + System.out.println("send failure"); + error.add(errorCode); + synchronized (lock) { + lock.notify(); + } + } + }); + lock.wait(30 * 1000); } - String sendData = "Conversation(" + getConvType() + ":" + getTarget() + "), Message(" + getMessage() + ")"; + result.sampleEnd(); - result.setSamplerData(sendData); - result.setSentBytes(sendData.length()); + result.setSamplerData(getMessage()); + result.setSentBytes(getMessage().length()); result.setLatency(result.getEndTime() - result.getStartTime()); - result.setSuccessful(pubCallback.isSuccessful()); + result.setSuccessful(!ret.isEmpty()); - if(pubCallback.isSuccessful()) { + if(result.isSuccessful()) { result.setResponseData("Publish successfuly.".getBytes()); - result.setResponseMessage(MessageFormat.format("publish successfully for Connection {0}.", connection)); + result.setResponseMessage(MessageFormat.format("publish successfully for Connection {0}.", imClient)); result.setResponseCodeOK(); } else { result.setSuccessful(false); - result.setResponseMessage(MessageFormat.format("Publish failed for connection {0}.", connection)); - result.setResponseData(("Publish failed with error code " + pubCallback.getErrorCode()).getBytes()); + result.setResponseMessage(MessageFormat.format("Publish failed for imClient {0}.", imClient)); + result.setResponseData("Publish failed with error code " + error.get(0)); result.setResponseCode("500"); } } catch (Exception ex) { @@ -260,7 +242,7 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis result.sampleEnd(); result.setLatency(result.getEndTime() - result.getStartTime()); result.setSuccessful(false); - result.setResponseMessage(MessageFormat.format("Publish failed for connection {0}.", connection)); + result.setResponseMessage(MessageFormat.format("Publish failed for imClient {0}.", imClient)); result.setResponseData(ex.getMessage().getBytes()); result.setResponseCode("500"); } @@ -275,13 +257,13 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis @Override public void threadFinished() { if (!isConnectionShare()) { - if (this.connection != null) { + if (this.imClient != null) { final CountDownLatch cdl = new CountDownLatch(1); Callback cb = new Callback() { @Override public void onSuccess(Void value) { - logger.info(MessageFormat.format("The connection {0} disconneted successfully.", connection)); + logger.info(MessageFormat.format("The imClient {0} disconneted successfully.", imClient)); cdl.countDown(); } @@ -292,7 +274,7 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis } }; logger.info("before disconnection"); - this.connection.disconnect(cb); + this.imClient.disconnect(true, cb); logger.info("after disconnection"); try { cdl.await(); @@ -300,11 +282,9 @@ public class SendMessageSampler extends AbstractMQTTSampler implements ThreadLis e.printStackTrace(); } } - if (ConnectionsManager.getInstance().containsConnection(connKey)) { - ConnectionsManager.getInstance().removeConnection(connKey); - } + ClientManager.getInstance().removeClient(clientKey); } else { - logger.info("share connection, not disconnect the connection"); + logger.info("share imClient, not disconnect the imClient"); } } public static String getUUID() {