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

broker/Client: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:43 +01:00
parent e42e1257c9
commit 42dca4a42e
1 changed files with 51 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/testclient/Client.java
+51 -44
View File
@@ -13,6 +13,7 @@
* *
* You may elect to redistribute this code under either of these licenses. * You may elect to redistribute this code under either of these licenses.
*/ */
package io.moquette.testclient; package io.moquette.testclient;
import io.moquette.BrokerConstants; import io.moquette.BrokerConstants;
@@ -25,22 +26,22 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*; import io.netty.handler.codec.mqtt.*;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/** /**
* Class used just to send and receive MQTT messages without any protocol login * Class used just to send and receive MQTT messages without any protocol login in action, just use
* in action, just use the encoder/decoder part. * the encoder/decoder part.
* *
* @author andrea * @author andrea
*/ */
public class Client { public class Client {
public interface ICallback { public interface ICallback {
void call(MqttMessage msg); void call(MqttMessage msg);
} }
private static final Logger LOG = LoggerFactory.getLogger(Client.class); private static final Logger LOG = LoggerFactory.getLogger(Client.class);
final ClientNettyMQTTHandler handler = new ClientNettyMQTTHandler(); final ClientNettyMQTTHandler handler = new ClientNettyMQTTHandler();
EventLoopGroup workerGroup; EventLoopGroup workerGroup;
@@ -49,11 +50,11 @@ public class Client {
private ICallback callback; private ICallback callback;
private String clientId; private String clientId;
private MqttMessage receivedMsg; private MqttMessage receivedMsg;
public Client(String host) { public Client(String host) {
this(host, BrokerConstants.PORT); this(host, BrokerConstants.PORT);
} }
public Client(String host, int port) { public Client(String host, int port) {
handler.setClient(this); handler.setClient(this);
workerGroup = new NioEventLoopGroup(); workerGroup = new NioEventLoopGroup();
@@ -63,6 +64,7 @@ public class Client {
b.channel(NioSocketChannel.class); b.channel(NioSocketChannel.class);
b.option(ChannelOption.SO_KEEPALIVE, true); b.option(ChannelOption.SO_KEEPALIVE, true);
b.handler(new ChannelInitializer<SocketChannel>() { b.handler(new ChannelInitializer<SocketChannel>() {
@Override @Override
public void initChannel(SocketChannel ch) throws Exception { public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline(); ChannelPipeline pipeline = ch.pipeline();
@@ -71,10 +73,10 @@ public class Client {
pipeline.addLast("handler", handler); pipeline.addLast("handler", handler);
} }
}); });
// Start the client. // Start the client.
m_channel = b.connect(host, port).sync().channel(); m_channel = b.connect(host, port).sync().channel();
} catch (Exception ex) { } catch (Exception ex) {
LOG.error("Error received in client setup", ex); LOG.error("Error received in client setup", ex);
workerGroup.shutdownGracefully(); workerGroup.shutdownGracefully();
} }
@@ -86,44 +88,49 @@ public class Client {
} }
public void connect(String willTestamentTopic, String willTestamentMsg) { public void connect(String willTestamentTopic, String willTestamentMsg) {
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0); MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
MqttConnectVariableHeader mqttConnectVariableHeader = MqttMessageType.CONNECT,
new MqttConnectVariableHeader( false,
MqttVersion.MQTT_3_1.protocolName(), MqttQoS.AT_MOST_ONCE,
MqttVersion.MQTT_3_1.protocolLevel(), false,
false, 0);
false, MqttConnectVariableHeader mqttConnectVariableHeader = new MqttConnectVariableHeader(
false, MqttVersion.MQTT_3_1.protocolName(),
MqttQoS.AT_MOST_ONCE.value(), MqttVersion.MQTT_3_1.protocolLevel(),
true, false,
true, false,
2); false,
MqttConnectPayload mqttConnectPayload = new MqttConnectPayload(this.clientId, willTestamentTopic, willTestamentMsg, MqttQoS.AT_MOST_ONCE.value(),
null, null); true,
MqttConnectMessage connectMessage = new MqttConnectMessage(mqttFixedHeader, mqttConnectVariableHeader, true,
2);
MqttConnectPayload mqttConnectPayload = new MqttConnectPayload(
this.clientId,
willTestamentTopic,
willTestamentMsg,
null,
null);
MqttConnectMessage connectMessage = new MqttConnectMessage(
mqttFixedHeader,
mqttConnectVariableHeader,
mqttConnectPayload); mqttConnectPayload);
/*
/*ConnectMessage connectMessage = new ConnectMessage(); * ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProtocolVersion((byte) 3); * connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(this.clientId);
connectMessage.setClientID(this.clientId); * connectMessage.setKeepAlive(2); //secs connectMessage.setWillFlag(true);
connectMessage.setKeepAlive(2); //secs * connectMessage.setWillMessage(willTestamentMsg.getBytes());
connectMessage.setWillFlag(true); * connectMessage.setWillTopic(willTestamentTopic);
connectMessage.setWillMessage(willTestamentMsg.getBytes()); * connectMessage.setWillQos(MqttQoS.AT_MOST_ONCE.byteValue());
connectMessage.setWillTopic(willTestamentTopic); */
connectMessage.setWillQos(MqttQoS.AT_MOST_ONCE.byteValue());*/
doConnect(connectMessage); doConnect(connectMessage);
} }
public void connect() { public void connect() {
MqttConnectMessage connectMessage = MessageBuilder.connect() MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1_1)
.protocolVersion(MqttVersion.MQTT_3_1_1) .clientId("").keepAlive(2) // secs
.clientId("") .willFlag(false).willQoS(MqttQoS.AT_MOST_ONCE).build();
.keepAlive(2) //secs
.willFlag(false)
.willQoS(MqttQoS.AT_MOST_ONCE)
.build();
doConnect(connectMessage); doConnect(connectMessage);
} }
@@ -154,7 +161,7 @@ public class Client {
public void setCallback(ICallback callback) { public void setCallback(ICallback callback) {
this.callback = callback; this.callback = callback;
} }
public void sendMessage(MqttMessage msg) { public void sendMessage(MqttMessage msg) {
m_channel.writeAndFlush(msg); m_channel.writeAndFlush(msg);
} }
@@ -169,11 +176,11 @@ public class Client {
this.callback.call(msg); this.callback.call(msg);
} }
} }
void setConnectionLost(boolean status) { void setConnectionLost(boolean status) {
m_connectionLost = status; m_connectionLost = status;
} }
public boolean isConnectionLost() { public boolean isConnectionLost() {
return m_connectionLost; return m_connectionLost;
} }