Notify packet sending listeners in new thread

sendListeners are now invoked *after* the packet has been put on the
wire.

Also sending listener exceptions are not catched and not only
NotConnectedExceptions. And a exception does not cause a 'break' but a
'continue' now. Log level is WARNING now.

smack-bosh/src/main/java/org/jivesoftware/smack/bosh/XMPPBOSHConnection.java

@@ -309,6 +309,9 @@ public class XMPPBOSHConnection extends AbstractXMPPConnection {
     private void sendElement(Element element) {
         try {
             send(ComposableBody.builder().setPayloadXML(element.toXML().toString()).build());
+            if (element instanceof Packet) {
+                firePacketSendingListeners((Packet) element);
+            }
         }
         catch (BOSHException e) {
             LOGGER.log(Level.SEVERE, "BOSHException in sendPacketInternal", e);

smack-core/src/main/java/org/jivesoftware/smack/AbstractXMPPConnection.java

@@ -63,6 +63,7 @@ import org.jivesoftware.smack.packet.PlainStreamElement;
 import org.jivesoftware.smack.provider.PacketExtensionProvider;
 import org.jivesoftware.smack.provider.ProviderManager;
 import org.jivesoftware.smack.rosterstore.RosterStore;
+import org.jivesoftware.smack.util.Async;
 import org.jivesoftware.smack.util.PacketParserUtils;
 import org.jxmpp.util.XmppStringUtils;
 import org.xmlpull.v1.XmlPullParser;
@@ -506,9 +507,6 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
         // the content of the packet.
         firePacketInterceptors(packet);
         sendPacketInternal(packet);
-        // Process packet writer listeners. Note that we're using the sending thread so it's
-        // expected that listeners are fast.
-        firePacketSendingListeners(packet);
     }
 
     @Override
@@ -703,11 +701,15 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
 
     /**
      * Process all packet listeners for sending packets.
+     * <p>
+     * Compared to {@link #firePacketInterceptors(Packet)}, the listeners will be invoked in a new thread.
+     * </p>
      * 
      * @param packet the packet to process.
      */
-    private void firePacketSendingListeners(Packet packet) {
+    @SuppressWarnings("javadoc")
-        List<PacketListener> listenersToNotify = new LinkedList<PacketListener>();
+    protected void firePacketSendingListeners(final Packet packet) {
+        final List<PacketListener> listenersToNotify = new LinkedList<PacketListener>();
         synchronized (sendListeners) {
             for (ListenerWrapper listenerWrapper : sendListeners.values()) {
                 if (listenerWrapper.filterMatches(packet)) {
@@ -715,15 +717,24 @@ public abstract class AbstractXMPPConnection implements XMPPConnection {
                 }
             }
         }
+        if (listenersToNotify.isEmpty()) {
+            return;
+        }
+        // Notify in a new thread, because we can
+        Async.go(new Runnable() {
+            @Override
+            public void run() {
                 for (PacketListener listener : listenersToNotify) {
                     try {
                         listener.processPacket(packet);
                     }
-            catch (NotConnectedException e) {
+                    catch (Exception e) {
-                LOGGER.log(Level.WARNING, "Got not connected exception, aborting");
+                        LOGGER.log(Level.WARNING, "Sending listener threw exception", e);
-                break;
+                        continue;
                     }
                 }
+            }}
+        , "Smack Sending Listeners Notification (" + getConnectionCounter() + ')');
     }
 
     @Override

smack-tcp/src/main/java/org/jivesoftware/smack/tcp/XMPPTCPConnection.java

@@ -1347,13 +1347,19 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
                 initalOpenStreamSend.reportSuccess();
                 // Write out packets from the queue.
                 while (!done()) {
-                    Element packet = nextStreamElement();
+                    Element element = nextStreamElement();
-                    if (packet != null) {
+                    if (element == null) {
+                        continue;
+                    }
+                    Packet packet = null;
+                    if (element instanceof Packet) {
+                        packet = (Packet) element;
+                    }
                     // Check if the stream element should be put to the unacknowledgedStanza
                     // queue. Note that we can not do the put() in sendPacketInternal() and the
                     // packet order is not stable at this point (sendPacketInternal() can be
                     // called concurrently).
-                        if (isSmEnabled() && packet instanceof Packet) {
+                    if (isSmEnabled() && packet != null) {
                         // If the unacknowledgedStanza queue is nearly full, request an new ack
                         // from the server in order to drain it
                         if (unacknowledgedStanzas.size() == 0.8 * XMPPTCPConnection.QUEUE_SIZE) {
@@ -1361,16 +1367,18 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
                             writer.flush();
                         }
                         try {
-                                unacknowledgedStanzas.put((Packet) packet);
+                            unacknowledgedStanzas.put(packet);
                         }
                         catch (InterruptedException e) {
                             throw new IllegalStateException(e);
                         }
                     }
-                        writer.write(packet.toXML().toString());
+                    writer.write(element.toXML().toString());
                     if (queue.isEmpty()) {
                         writer.flush();
                     }
+                    if (packet != null) {
+                        firePacketSendingListeners(packet);
                     }
                 }
                 if (!instantShutdown) {