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

Java Version set to 1.8

Browse Source 
anonymous types replaced by lambda's
Method references used instead of Lambda's
Explicit type replaced by <>
travis.yml updated to java 1.8
This commit is contained in:
Andreas Mager
2017-06-19 17:07:05 +02:00
committed by Andreas Mager
parent 555893d287
commit fb9765304c
11 changed files with 32 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/Server.java
+2 -8
View File
@@ -75,14 +75,8 @@ public class Server {
final Server server = new Server();
server.startServer();
System.out.println("Server started, version 0.10-SNAPSHOT");
// Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
server.stopServer();
}
});
//Bind a shutdown hook
Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
}
/**
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+7 -49
View File
@@ -94,13 +94,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) {
LOG.debug("Sending MQTT CONNECT message to interceptor. CId={}, interceptorId={}",
msg.payload().clientIdentifier(), handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onConnect(new InterceptConnectMessage(msg));
}
});
executor.execute(() -> handler.onConnect(new InterceptConnectMessage(msg)));
}
}
@@ -109,13 +103,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) {
LOG.debug("Notifying MQTT client disconnection to interceptor. CId={}, username={}, interceptorId={}",
clientID, username, handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onDisconnect(new InterceptDisconnectMessage(clientID, username));
}
});
executor.execute(() -> handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)));
}
}
@@ -124,13 +112,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) {
LOG.debug("Notifying unexpected MQTT client disconnection to interceptor CId={}, username={}, " +
"interceptorId={}", clientID, username, handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username));
}
});
executor.execute(() -> handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)));
}
}
@@ -141,13 +123,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) {
LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}",
clientID, messageId, topic, handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onPublish(new InterceptPublishMessage(msg, clientID, username));
}
});
executor.execute(() -> handler.onPublish(new InterceptPublishMessage(msg, clientID, username)));
}
}
@@ -156,13 +132,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) {
LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. CId={}, topicFilter={}, interceptorId={}",
sub.getClientId(), sub.getTopicFilter(), handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onSubscribe(new InterceptSubscribeMessage(sub, username));
}
});
executor.execute(() -> handler.onSubscribe(new InterceptSubscribeMessage(sub, username)));
}
}
@@ -171,13 +141,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) {
LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. CId={}, topic={}, interceptorId={}", clientID,
topic, handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username));
}
});
executor.execute(() -> handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)));
}
}
@@ -186,13 +150,7 @@ final class BrokerInterceptor implements Interceptor {
for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) {
LOG.debug("Notifying MQTT ACK message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}",
msg.getMsg().getClientID(), msg.getPacketID(), msg.getTopic(), handler.getID());
executor.execute(new Runnable() {
@Override
public void run() {
handler.onMessageAcknowledged(msg);
}
});
executor.execute(() -> handler.onMessageAcknowledged(msg));
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+6 -9
View File
@@ -822,8 +822,10 @@ public class ProtocolProcessor {
if (req.qualityOfService() == FAILURE) {
continue;
}
Subscription newSubscription =
new Subscription(clientID, new Topic(req.topicName()), req.qualityOfService());
Subscription newSubscription = new Subscription(
clientID,
new Topic(req.topicName()),
req.qualityOfService());
clientSession.subscribe(newSubscription);
newSubscriptions.add(newSubscription);
@@ -889,13 +891,8 @@ public class ProtocolProcessor {
// scans retained messages to be published to the new subscription
// TODO this is ugly, it does a linear scan on potential big dataset
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
@Override
public boolean match(Topic key) {
return key.match(newSubscription.getTopicFilter());
}
});
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
.searchMatching(key -> key.match(newSubscription.getTopicFilter()));
if (!messages.isEmpty()) {
LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}",
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+3 -13
View File
@@ -19,7 +19,6 @@ package io.moquette.spi.impl;
import io.moquette.interception.InterceptHandler;
import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
@@ -417,12 +416,8 @@ public class ProtocolProcessorTest {
.payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).messageId(100).build();
m_processor.processPublish(m_channel, msg);
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(Topic key) {
return key.match(new Topic(FAKE_TOPIC));
}
});
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
.searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
assertFalse(messages.isEmpty());
// Exercise
@@ -432,12 +427,7 @@ public class ProtocolProcessorTest {
m_processor.processPublish(m_channel, cleanPubMsg);
// Verify
messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(Topic key) {
return key.match(new Topic(FAKE_TOPIC));
}
});
messages = m_messagesStore.searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
assertTrue(messages.isEmpty());
}
broker/src/test/java/io/moquette/testclient/Client.java
+3 -6
View File
@@ -134,12 +134,9 @@ public class Client {
private void doConnect(MqttConnectMessage connectMessage) {
final CountDownLatch latch = new CountDownLatch(1);
this.setCallback(new Client.ICallback() {
public void call(MqttMessage msg) {
receivedMsg = msg;
latch.countDown();
}
this.setCallback(msg -> {
receivedMsg = msg;
latch.countDown();
});
this.sendMessage(connectMessage);