Merge pull request #264 from AndreasMager/java-1.8

Java 1.8

.travis.yml

@@ -1 +1,3 @@
 language: java
+jdk:
+  - oraclejdk8

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
+        //Bind  a shutdown hook
-        Runtime.getRuntime().addShutdownHook(new Thread() {
+        Runtime.getRuntime().addShutdownHook(new Thread(server::stopServer));
-
-            @Override
-            public void run() {
-                server.stopServer();
-            }
-        });
     }
 
     /**

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() {
+            executor.execute(() -> handler.onConnect(new InterceptConnectMessage(msg)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onPublish(new InterceptPublishMessage(msg, clientID, username)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onSubscribe(new InterceptSubscribeMessage(sub, username)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)));
-
-                @Override
-                public void run() {
-                    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() {
+            executor.execute(() -> handler.onMessageAcknowledged(msg));
-
-                @Override
-                public void run() {
-                    handler.onMessageAcknowledged(msg);
-                }
-            });
         }
     }
 

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -822,8 +822,10 @@ public class ProtocolProcessor {
             if (req.qualityOfService() == FAILURE) {
                 continue;
             }
-            Subscription newSubscription =
+            Subscription newSubscription = new Subscription(
-                    new Subscription(clientID, new Topic(req.topicName()), req.qualityOfService());
+                    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() {
+        Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
-
+                .searchMatching(key -> key.match(newSubscription.getTopicFilter()));
-            @Override
-            public boolean match(Topic key) {
-                return 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

@@ -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() {
+        Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
-
+                .searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
-            public boolean match(Topic key) {
-                return 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() {
+        messages = m_messagesStore.searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
-
-            public boolean match(Topic key) {
-                return key.match(new Topic(FAKE_TOPIC));
-            }
-        });
         assertTrue(messages.isEmpty());
     }
 

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() {
+        this.setCallback(msg -> {
-
+            receivedMsg = msg;
-            public void call(MqttMessage msg) {
+            latch.countDown();
-                receivedMsg = msg;
-                latch.countDown();
-            }
         });
 
         this.sendMessage(connectMessage);

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() {
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-            @Override
+            System.out.println("Stopping broker");
-            public void run() {
+            mqttBroker.stopServer();
-                System.out.println("Stopping broker");
+            System.out.println("Broker stopped");
-                mqttBroker.stopServer();
+        }));
-                System.out.println("Broker stopped");
-            }
-        });
 
         Thread.sleep(20000);
         System.out.println("Before self publish");

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() {
+        m_scheduler.scheduleWithFixedDelay(() -> {
-
-            @Override
-            public void run() {
                 LOG.debug("Committing to MapDB");
                 m_db.commit();
-            }
         }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS);
 
         // TODO check m_db is valid and

mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java

@@ -191,7 +191,7 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore {
         Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
         if (inFlightForClient == null) {
             int nextPacketId = 1;
-            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
+            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>());
             inFlightForClient.add(nextPacketId);
             this.m_inFlightIds.put(clientID, inFlightForClient);
             return nextPacketId;

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() {
+        Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
-            @Override
-            public void run() {
-                server.stop();
-            }
-        });
     }
 
     private static void startClientsTesting() throws MqttException, InterruptedException {

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() {
+        Runtime.getRuntime().addShutdownHook(new Thread(server::stop));
-            @Override
-            public void run() {
-                server.stop();
-            }
-        });
     }
 
     private static void publishBombing() {