Handle publish errors properly in PersistentQueueMessageSender. Create a connection descriptor store to handle all the message delivery errors and to avoid

duplicating error handling code (i.e. the connection descriptors map is null, the connection descriptor does not exist, etc).

broker/src/main/java/io/moquette/server/ConnectionDescriptor.java

@@ -16,10 +16,16 @@
 package io.moquette.server;
 
 import io.netty.channel.Channel;
+import java.util.NoSuchElementException;
+import java.util.concurrent.TimeUnit;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.atomic.AtomicReference;
+import io.moquette.server.netty.AutoFlushHandler;
+import io.moquette.server.netty.NettyUtils;
+
 
 /**
  * Value object to maintain the information of single connection, like ClientID, Channel,
@@ -40,7 +46,7 @@ public class ConnectionDescriptor {
     }
 
     public final String clientID;
-    public final Channel channel;
+    private final Channel channel;
     public final boolean cleanSession;
     private final AtomicReference<ConnectionState> channelState = new AtomicReference<>(ConnectionState.DISCONNECTED);
 
@@ -49,20 +55,59 @@ public class ConnectionDescriptor {
         this.channel = session;
         this.cleanSession = cleanSession;
     }
+    
+    public void writeAndFlush(Object payload) {
+    	this.channel.writeAndFlush(payload);
+    }
+    
+    public void setupAutoFlusher(int flushIntervalMs) {
+        try {
+            this.channel.pipeline().addAfter("idleEventHandler", "autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
+        } catch (NoSuchElementException nseex) {
+            //the idleEventHandler is not present on the pipeline
+            this.channel.pipeline().addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
+        }
+    }
+    
+    public boolean doesNotUseChannel(Channel channel) {
+    	return !(this.channel.equals(channel));
+    }
+    
+	public boolean close() {
+		LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
+		final boolean success = assignState(ConnectionState.INTERCEPTORS_NOTIFIED, ConnectionState.DISCONNECTED);
+		if (!success) {
+			return false;
+		}
+		this.channel.close();
+		return true;
+	}
+	
+    public String getUsername() {
+    	return NettyUtils.userName(this.channel);
+    }
 
     public void abort() {
-        LOG.info("closing the channel");
+    	LOG.info("Closing connection descriptor. MqttClientId = {}.", clientID);
 //        try {
             //this.channel.disconnect().sync();
-            this.channel.close();//.sync();
+        this.channel.close();//.sync();
 //        } catch (InterruptedException e) {
 //            e.printStackTrace();
 //        }
     }
 
     public boolean assignState(ConnectionState expected, ConnectionState newState) {
-        return channelState.compareAndSet(expected, newState);
-    }
+		LOG.debug("Updating state of connection descriptor. MqttClientId = {}, expectedState = {}, newState = {}.",
+				clientID, expected, newState);
+		boolean retval = channelState.compareAndSet(expected, newState);
+		if (!retval) {
+			LOG.error(
+					"Unable to update state of connection descriptor. MqttclientId = {}, expectedState = {}, newState = {}.",
+					clientID, expected, newState);
+		}
+		return retval;
+	}
 
     @Override
     public String toString() {

broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java

@@ -0,0 +1,85 @@
+package io.moquette.server;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.moquette.parser.proto.messages.AbstractMessage;
+
+public class ConnectionDescriptorStore {
+
+	private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class);
+
+	private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
+
+	public ConnectionDescriptorStore() {
+		this.connectionDescriptors = new ConcurrentHashMap<String, ConnectionDescriptor>();
+	}
+
+	public boolean sendMessage(AbstractMessage message, String clientId) {
+		return sendMessage(message, null, clientId);
+	}
+
+	public boolean sendMessage(AbstractMessage message, Integer messageID, String clientID) {
+		try {
+			// The connection descriptors map will never be null: it's a final
+			// attribute.
+
+			if (messageID != null) {
+				LOG.info("Sending {} message. MqttClientId = {}, messageId = {}.", message.getMessageType(), clientID,
+						messageID);
+			} else {
+				LOG.debug("Sending {} message. MqttClientId = {}.", message.getMessageType(), clientID);
+			}
+
+			ConnectionDescriptor descriptor = connectionDescriptors.get(clientID);
+			if (descriptor == null) {
+				if (messageID != null) {
+					LOG.error(
+							"The client has just disconnected. The {} message could not be sent. MqttClientId = {}, messageId = {}.",
+							message.getMessageType(), clientID, messageID);
+				} else {
+					LOG.error("The client has just disconnected. The {} could not be sent. MqttClientId = {}.",
+							message.getMessageType(), clientID);
+				}
+				/*
+				 * If the client hast just disconnected, its connection
+				 * descriptor will be null. We don't have to make the broker
+				 * crash: we'll just discard the PUBACK message.
+				 */
+				return false;
+			}
+			descriptor.writeAndFlush(message);
+			return true;
+		} catch (Throwable e) {
+			if (messageID != null) {
+				LOG.error(
+						"Unable to send {} message. MqttClientId = {}, messageId = {}, cause = {}, errorMessage = {}.",
+						message.getMessageType(), clientID, messageID, e.getCause(), e.getMessage());
+			} else {
+				LOG.error("Unable to send {} message. MqttClientId = {}, cause = {}, errorMessage = {}.",
+						message.getMessageType(), clientID, e.getCause(), e.getMessage());
+			}
+			return false;
+		}
+	}
+
+	public ConnectionDescriptor addConnection(ConnectionDescriptor descriptor) {
+		return connectionDescriptors.putIfAbsent(descriptor.clientID, descriptor);
+	}
+
+	public boolean removeConnection(ConnectionDescriptor descriptor) {
+		return connectionDescriptors.remove(descriptor.clientID, descriptor);
+	}
+
+	public ConnectionDescriptor getConnection(String clientID) {
+		return connectionDescriptors.get(clientID);
+	}
+	
+	public boolean isConnected(String clientID) {
+		return connectionDescriptors.containsKey(clientID);
+	}
+
+}