Repository: moquette-io/moquette
Issue: NullPointerException caused by async connect/disconnect processing

### Expected behavior
Session is not removed by handleConnectionLost if new session is already reopened

### Actual behavior
1. handleConnectionLost starts
2. processConnect starts and reopen session
3. handleConnectionLost continues and removes session
4. PostOffice.flushInFlight fails to use session

### Steps to reproduce
1. Run broker
2. Use clean session=true in mqtt-client
3. Start and stop mqtt-client several times. Try to do pause between stop and start as short as possible.
P.S. It seems like one more mqtt-client connected permanently gives better chance to reproduce.

### Minimal yet complete reproducer code (or URL to code) or complete log file
I added extra logging to catch the problem
```
14:59:39.254 [nioEventLoopGroup-3-6] io.moquette.broker.SessionRegistry - retrieve session OK by handleConnectionLost

14:59:39.254 [nioEventLoopGroup-3-7] io.moquette.broker.MQTTConnection  - processConnect started
14:59:39.254 [nioEventLoopGroup-3-7] io.moquette.broker.SessionRegistry - NEW SESSION
14:59:39.255 [nioEventLoopGroup-3-7] io.moquette.broker.SessionRegistry - REOPEN SESSION

14:59:39.255 [nioEventLoopGroup-3-6] io.moquette.broker.SessionRegistry - remove session by handleConnectionLost

14:59:39.255 [nioEventLoopGroup-3-7] io.moquette.broker.MQTTConnection  - processConnect ACK started
14:59:39.257 [nioEventLoopGroup-3-7] io.moquette.broker.MQTTConnection  - processConnect ACK end with true
14:59:39.257 [nioEventLoopGroup-3-7] io.moquette.broker.SessionRegistry - retrieve session NULL by flushInFlight

14:59:39.257 [nioEventLoopGroup-3-7] io.moquette.broker.NewNettyMQTTHandler  - Unexpected exception while processing MQTT message. Closing Netty channel. CId=
 java.lang.NullPointerException: Attempt to invoke virtual method 'void io.moquette.broker.Session.flushAllQueuedMessages()' on a null object reference
     at io.moquette.broker.PostOffice.flushInFlight(PostOffice.java:319)
     at io.moquette.broker.MQTTConnection.readCompleted(MQTTConnection.java:536)
     at io.moquette.broker.NewNettyMQTTHandler.channelReadComplete(NewNettyMQTTHandler.java:76)
```
### Moquette MQTT version
0.12.1, master 4665c5e

### JVM version (e.g. `java -version`)
1.8

Changed files:
broker/src/main/java/io/moquette/broker/SessionRegistry.java

diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
index a65cdc96..d86cedb7 100644
--- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java
+++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java
@@ -90,9 +90,10 @@ public class SessionRegistry {
 
     SessionCreationResult createOrReopenSession(MqttConnectMessage msg, String clientId, String username) {
         SessionCreationResult postConnectAction;
-        if (!pool.containsKey(clientId)) {
+        final Session newSession = createNewSession(msg, clientId);
+        final Session oldSession = pool.get(clientId);
+        if (oldSession == null) {
             // case 1
-            final Session newSession = createNewSession(msg, clientId);
             postConnectAction = new SessionCreationResult(newSession, CreationModeEnum.CREATED_CLEAN_NEW, false);
 
             // publish the session
@@ -102,19 +103,17 @@ public class SessionRegistry {
             if (success) {
                 LOG.trace("case 1, not existing session with CId {}", clientId);
             } else {
-                postConnectAction = reopenExistingSession(msg, clientId, newSession, username);
+                postConnectAction = reopenExistingSession(msg, clientId, previous, newSession, username);
             }
         } else {
-            final Session newSession = createNewSession(msg, clientId);
-            postConnectAction = reopenExistingSession(msg, clientId, newSession, username);
+            postConnectAction = reopenExistingSession(msg, clientId, oldSession, newSession, username);
         }
         return postConnectAction;
     }
 
     private SessionCreationResult reopenExistingSession(MqttConnectMessage msg, String clientId,
-                                                        Session newSession, String username) {
+                                                        Session oldSession, Session newSession, String username) {
         final boolean newIsClean = msg.variableHeader().isCleanSession();
-        final Session oldSession = pool.get(clientId);
         final SessionCreationResult creationResult;
         if (oldSession.disconnected()) {
             if (newIsClean) {
@@ -150,16 +149,17 @@ public class SessionRegistry {
             creationResult = new SessionCreationResult(newSession, CreationModeEnum.DROP_EXISTING, true);
         }
 
-        final boolean published;
         if (creationResult.mode == CreationModeEnum.DROP_EXISTING) {
             LOG.debug("Drop session of already connected client with same id");
-            published = pool.replace(clientId, oldSession, newSession);
+            if (!pool.replace(clientId, oldSession, newSession)) {
+                //the other client was disconnecting and removed it's own session
+                pool.put(clientId, newSession);
+            }
         } else {
             LOG.debug("Replace session of client with same id");
-            published = pool.replace(clientId, oldSession, oldSession);
-        }
-        if (!published) {
-            throw new SessionCorruptedException("old session was already removed");
+            if (!pool.replace(clientId, oldSession, oldSession)) {
+                throw new SessionCorruptedException("old session was already removed");
+            }
         }
 
         // case not covered new session is clean true/false and old session not in CONNECTED/DISCONNECTED