From fb9765304c0c11c1aeaad17ca53fcf812a7329fa Mon Sep 17 00:00:00 2001 From: Andreas Mager Date: Thu, 12 Jan 2017 20:39:44 +0100 Subject: [PATCH] Java Version set to 1.8 anonymous types replaced by lambda's Method references used instead of Lambda's Explicit type replaced by <> travis.yml updated to java 1.8 --- .travis.yml | 2 + .../main/java/io/moquette/server/Server.java | 10 +--- .../moquette/spi/impl/BrokerInterceptor.java | 56 +++---------------- .../moquette/spi/impl/ProtocolProcessor.java | 15 ++--- .../spi/impl/ProtocolProcessorTest.java | 16 +----- .../java/io/moquette/testclient/Client.java | 9 +-- .../testembedded/EmbeddedLauncher.java | 13 ++--- .../mapdb/MapDBPersistentStore.java | 6 +- .../persistence/mapdb/MapDBSessionsStore.java | 2 +- .../performance/ProtocolDecodingServer.java | 7 +-- .../ProtocolPublishDecodingServer.java | 7 +-- 11 files changed, 32 insertions(+), 111 deletions(-) diff --git a/.travis.yml b/.travis.yml index dff5f3a5..9bcf9994 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1 +1,3 @@ language: java +jdk: + - oraclejdk8 diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index b4a690c8..37b732a9 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -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)); } /** diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 2acd3ede..61d0ca38 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -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)); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 5eabfe81..97ddff34 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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 messages = m_messagesStore.searchMatching(new IMatchingCondition() { - - @Override - public boolean match(Topic key) { - return key.match(newSubscription.getTopicFilter()); - } - }); + Collection messages = m_messagesStore + .searchMatching(key -> key.match(newSubscription.getTopicFilter())); if (!messages.isEmpty()) { LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}", diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 02eae29e..589761a3 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -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 messages = m_messagesStore.searchMatching(new IMatchingCondition() { - - public boolean match(Topic key) { - return key.match(new Topic(FAKE_TOPIC)); - } - }); + Collection 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()); } diff --git a/broker/src/test/java/io/moquette/testclient/Client.java b/broker/src/test/java/io/moquette/testclient/Client.java index 8930ded2..52c35d4f 100644 --- a/broker/src/test/java/io/moquette/testclient/Client.java +++ b/broker/src/test/java/io/moquette/testclient/Client.java @@ -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); diff --git a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java index 0b0bf1db..6a4c2a30 100644 --- a/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java +++ b/embedding_moquette/src/main/java/io/moquette/testembedded/EmbeddedLauncher.java @@ -59,14 +59,11 @@ final class EmbeddedLauncher { System.out.println("Broker started press [CTRL+C] to stop"); //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread() { - @Override - public void run() { - System.out.println("Stopping broker"); - mqttBroker.stopServer(); - System.out.println("Broker stopped"); - } - }); + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + System.out.println("Stopping broker"); + mqttBroker.stopServer(); + System.out.println("Broker stopped"); + })); Thread.sleep(20000); System.out.println("Before self publish"); diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java index 0cb0b086..7932d59b 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java @@ -83,13 +83,9 @@ public class MapDBPersistentStore implements IStore { m_db = DBMaker.newFileDB(tmpFile).make(); } LOG.info("Scheduling MapDB commit task"); - m_scheduler.scheduleWithFixedDelay(new Runnable() { - - @Override - public void run() { + m_scheduler.scheduleWithFixedDelay(() -> { LOG.debug("Committing to MapDB"); m_db.commit(); - } }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); // TODO check m_db is valid and diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 057d8474..21aa99b3 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -191,7 +191,7 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { Set inFlightForClient = this.m_inFlightIds.get(clientID); if (inFlightForClient == null) { int nextPacketId = 1; - inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap()); + inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>()); inFlightForClient.add(nextPacketId); this.m_inFlightIds.put(clientID, inFlightForClient); return nextPacketId; diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java index c2dbbdf3..d0371c4a 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolDecodingServer.java @@ -152,12 +152,7 @@ public class ProtocolDecodingServer { System.out.println("Loop server started"); startClientsTesting(); //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread() { - @Override - public void run() { - server.stop(); - } - }); + Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); } private static void startClientsTesting() throws MqttException, InterruptedException { diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java index 2804545a..6d2ec1eb 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/ProtocolPublishDecodingServer.java @@ -104,12 +104,7 @@ public class ProtocolPublishDecodingServer { publishBombing(); publishBombing(); //Bind a shutdown hook - Runtime.getRuntime().addShutdownHook(new Thread() { - @Override - public void run() { - server.stop(); - } - }); + Runtime.getRuntime().addShutdownHook(new Thread(server::stop)); } private static void publishBombing() {