1. Made asynchronic packets auditing. JM-25

2. Fixed bug while starting up AuditManager.


git-svn-id: http://svn.igniterealtime.org/svn/repos/messenger/trunk@435 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/messenger/audit/AuditManager.java

@@ -88,6 +88,24 @@ public interface AuditManager {
      */
     void setMaxFileCount(int count);
 
+    /**
+     * Returns the time in milliseconds between successive executions of the task that will save
+     * the queued audited packets to a permanent store.
+     *
+     * @return the time in milliseconds between successive executions of the task that will save
+     *         the queued audited packets to a permanent store.
+     */
+    int getLogTimeout();
+
+    /**
+     * Sets the time in milliseconds between successive executions of the task that will save
+     * the queued audited packets to a permanent store.
+     *
+     * @param logTimeout the time in milliseconds between successive executions of the task that will save
+     *        the queued audited packets to a permanent store. 
+     */
+    void setLogTimeout(int logTimeout);
+
     /**
      * <p>Determines if the server will audit all message packets.</p>
      * <p>This is a speed optimization and convenience for logging all message packets

src/java/org/jivesoftware/messenger/audit/Auditor.java

@@ -11,9 +11,6 @@
 
 package org.jivesoftware.messenger.audit;
 
-import org.jivesoftware.messenger.IQ;
-import org.jivesoftware.messenger.Message;
-import org.jivesoftware.messenger.Presence;
 import org.jivesoftware.messenger.XMPPPacket;
 
 /**
@@ -34,44 +31,30 @@ public interface Auditor {
      */
     void audit(XMPPPacket packet);
 
-    /**
-     * Audit a message packet.
-     *
-     * @param packet the packet being audited.
-     */
-    void audit(Message packet);
-
-    /**
-     * Audit a presence packet.
-     *
-     * @param packet the packet being audited.
-     * @param transition the presence transition type from AuditManager.
-     */
-    void audit(Presence packet, int transition);
-
-    /**
-     * Audit an IQ packet.
-     *
-     * @param packet the packet being audited.
-     */
-    void audit(IQ packet);
-
     /**
      * Audit any packet that was dropped (undeliverables, etc).
      *
      * @param packet the packet that was dropped.
      */
-    void auditDroppedPacket(XMPPPacket packet);
+    //void auditDroppedPacket(XMPPPacket packet);
 
     /**
      * Audit a non-packet event.
      *
      * @param event the event being audited.
      */
-    void audit(AuditEvent event);
+    //void audit(AuditEvent event);
 
     /**
      * Prepares the auditor for system shutdown.
      */
-    void close();
+    void stop();
+
+    /**
+     * Returns the number of queued packets that are still in memory and need to be saved to a
+     * permanent store.
+     *
+     * @return the number of queued packets that are still in memory.
+     */
+    int getQueuedPacketsNumber();
 }
\ No newline at end of file

src/java/org/jivesoftware/messenger/audit/spi/AuditManagerImpl.java

@@ -30,8 +30,10 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
     private AuditorImpl auditor = null;
     private int maxSize;
     private int maxCount;
+    private int logTimeout;
     private static final int MAX_FILE_SIZE = 10;
     private static final int MAX_FILE_COUNT = 10;
+    private static final int DEFAULT_LOG_TIMEOUT = 120000;
 
     public AuditManagerImpl() {
         super("Audit Manager");
@@ -63,6 +65,16 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
         JiveGlobals.setProperty("xmpp.audit.maxsize", Integer.toString(size));
     }
 
+    public int getLogTimeout() {
+        return logTimeout;
+    }
+
+    public void setLogTimeout(int logTimeout) {
+        this.logTimeout = logTimeout;
+        auditor.setLogTimeout(logTimeout);
+        JiveGlobals.setProperty("xmpp.audit.logtimeout", Integer.toString(logTimeout));
+    }
+
     public int getMaxFileCount() {
         return maxCount;
     }
@@ -135,11 +147,11 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
 
     public void initialize(Container container) {
         super.initialize(container);
-        setEnabled(JiveGlobals.getBooleanProperty("xmpp.audit.active"));
-        setEnabled(JiveGlobals.getBooleanProperty("xmpp.audit.message"));
-        setEnabled(JiveGlobals.getBooleanProperty("xmpp.audit.presence"));
-        setEnabled(JiveGlobals.getBooleanProperty("xmpp.audit.iq"));
-        setEnabled(JiveGlobals.getBooleanProperty("xmpp.audit.xpath"));
+        enabled = JiveGlobals.getBooleanProperty("xmpp.audit.active");
+        auditMessage = JiveGlobals.getBooleanProperty("xmpp.audit.message");
+        auditPresence = JiveGlobals.getBooleanProperty("xmpp.audit.presence");
+        auditIQ = JiveGlobals.getBooleanProperty("xmpp.audit.iq");
+        auditXPath = JiveGlobals.getBooleanProperty("xmpp.audit.xpath");
         // TODO: load xpath values!
 //        String[] filters = context.getProperties("xmpp.audit.filter.xpath");
 //        for (int i = 0; i < filters.length; i++) {
@@ -147,13 +159,15 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
 //        }
         maxSize = JiveGlobals.getIntProperty("xmpp.audit.maxsize", MAX_FILE_SIZE);
         maxCount = JiveGlobals.getIntProperty("xmpp.audit.maxcount", MAX_FILE_COUNT);
+        logTimeout = JiveGlobals.getIntProperty("xmpp.audit.logtimeout", DEFAULT_LOG_TIMEOUT);
         auditor = new AuditorImpl(this);
         auditor.setMaxValues(maxSize, maxCount);
+        auditor.setLogTimeout(logTimeout);
     }
 
     public void stop() {
         if (auditor != null) {
-            auditor.close();
+            auditor.stop();
         }
     }
 }

src/java/org/jivesoftware/messenger/audit/spi/AuditorImpl.java

@@ -22,9 +22,13 @@ import java.io.FileWriter;
 import java.io.IOException;
 import java.io.Writer;
 import java.util.Date;
+import java.util.Queue;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import javax.xml.stream.XMLStreamWriter;
 import javax.xml.stream.XMLOutputFactory;
 import javax.xml.stream.XMLStreamException;
-import javax.xml.stream.XMLStreamWriter;
 
 public class AuditorImpl implements Auditor {
 
@@ -35,8 +39,20 @@ public class AuditorImpl implements Auditor {
     private static final int MEGABYTE = 1024 * 1024;
     private int maxSize;
     private long maxCount;
+    private int logTimeout;
     private boolean closed = false;
 
+    /**
+     * Queue that holds the audited packets that will be later saved to an XML file.
+     */
+    private Queue<AuditPacket> logQueue = new LinkedBlockingQueue<AuditPacket>();
+
+    /**
+     * Timer to save queued logs to the XML file.
+     */
+    private Timer timer = new Timer();
+    private SaveQueuedPacketsTask saveQueuedPacketsTask;
+
     public AuditorImpl(AuditManager manager) {
         auditManager = manager;
     }
@@ -44,55 +60,40 @@ public class AuditorImpl implements Auditor {
     public void audit(XMPPPacket packet) {
         if (auditManager.isEnabled()) {
             if (packet instanceof Message) {
-                if (auditManager.isEnabled() && auditManager.isAuditMessage()) {
+                if (auditManager.isAuditMessage()) {
                     writePacket(packet, false);
                 }
             }
             else if (packet instanceof Presence) {
-                if (auditManager.isEnabled() && auditManager.isAuditPresence()) {
+                if (auditManager.isAuditPresence()) {
                     writePacket(packet, false);
                 }
             }
             else if (packet instanceof IQ) {
-                if (auditManager.isEnabled() && auditManager.isAuditIQ()) {
+                if (auditManager.isAuditIQ()) {
                     writePacket(packet, false);
                 }
             }
         }
     }
 
-    public synchronized void audit(Message packet) {
-        if (auditManager.isEnabled() && auditManager.isAuditMessage()) {
-            writePacket(packet, false);
-        }
-    }
-
-    public synchronized void audit(Presence packet, int transition) {
-        if (auditManager.isEnabled() && auditManager.isAuditPresence()) {
-            writePacket(packet, false);
-        }
-    }
-
-    public synchronized void audit(IQ packet) {
-        if (auditManager.isEnabled() && auditManager.isAuditIQ()) {
-            writePacket(packet, false);
-        }
-    }
-
-    public synchronized void auditDroppedPacket(XMPPPacket packet) {
+    /*public void auditDroppedPacket(XMPPPacket packet) {
         writePacket(packet, true);
     }
 
-    public synchronized void audit(AuditEvent event) {
-        try {
-            prepareAuditFile();
-        }
-        catch (Exception e) {
-            Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
-        }
+    public void audit(AuditEvent event) {
+        // TODO Implement this functionality. Not used currently.
+    }*/
+
+    public void stop() {
+        // Stop the scheduled task for saving queued packets to the XML file
+        timer.cancel();
+        // Save all remaining queued packets to the XML file
+        saveQueuedPackets();
+        close();
     }
 
-    public void close() {
+    private void close() {
         if (xmlSerializer != null) {
             try {
                 xmlSerializer.writeEndElement();
@@ -109,48 +110,9 @@ public class AuditorImpl implements Auditor {
 
     private void writePacket(XMPPPacket packet, boolean dropped) {
         if (!closed) {
-            try {
-                prepareAuditFile();
-                xmlSerializer.writeStartElement("packet");
-                xmlSerializer.writeDefaultNamespace("http://jivesoftware.org");
-                Session session = packet.getOriginatingSession();
-                if (session != null) {
-                    if (session.getStreamID() != null) {
-                        xmlSerializer.writeAttribute("session", session.getStreamID().toString());
-                    }
-                    switch (session.getStatus()) {
-                        case Session.STATUS_AUTHENTICATED:
-                            xmlSerializer.writeAttribute("status", "auth");
-                            break;
-                        case Session.STATUS_CLOSED:
-                            xmlSerializer.writeAttribute("status", "closed");
-                            break;
-                        case Session.STATUS_CONNECTED:
-                            xmlSerializer.writeAttribute("status", "connected");
-                            break;
-                        case Session.STATUS_STREAMING:
-                            xmlSerializer.writeAttribute("status", "stream");
-                            break;
-                        default:
-                            xmlSerializer.writeAttribute("status", "unknown");
-                            break;
-                    }
-                }
-                xmlSerializer.writeAttribute("timestamp", new Date().toString());
-                if (packet.isSending()) {
-                    xmlSerializer.writeAttribute("sending", "true");
-                }
-                if (dropped) {
-                    xmlSerializer.writeAttribute("dropped", "true");
-                }
-                packet.send(xmlSerializer, 0);
-                xmlSerializer.writeEndElement();
-                xmlSerializer.flush();
-            }
-            catch (Exception e) {
-                Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
-            }
-        } // closed
+            // Add to the logging queue this new entry that will be saved later
+            logQueue.add(new AuditPacket((XMPPPacket) packet.createDeepCopy(), dropped));
+        }
     }
 
     private void prepareAuditFile() throws IOException, XMLStreamException {
@@ -164,6 +126,22 @@ public class AuditorImpl implements Auditor {
         maxCount = count;
     }
 
+    public void setLogTimeout(int newTimeout) {
+        // Cancel any existing task because the timeout has changed
+        if (saveQueuedPacketsTask != null) {
+            saveQueuedPacketsTask.cancel();
+        }
+        this.logTimeout = newTimeout;
+        // Create a new task and schedule it with the new timeout
+        saveQueuedPacketsTask = new SaveQueuedPacketsTask();
+        timer.schedule(saveQueuedPacketsTask, logTimeout, logTimeout);
+
+    }
+
+    public int getQueuedPacketsNumber() {
+        return logQueue.size();
+    }
+
     private void rotateFiles() throws IOException, XMLStreamException {
         close();
         int i;
@@ -203,4 +181,117 @@ public class AuditorImpl implements Auditor {
         xmlSerializer.writeStartElement("jive", "jive", "http://jivesoftware.org");
         xmlSerializer.writeNamespace("jive", "http://jivesoftware.org");
     }
+
+    /**
+     * Saves the queued entries to an XML file.
+     */
+    private class SaveQueuedPacketsTask extends TimerTask {
+        public void run() {
+            try {
+                saveQueuedPackets();
+            }
+            catch (Throwable e) {
+                Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+            }
+        }
+    }
+
+    private void saveQueuedPackets() {
+        AuditPacket entry;
+        int batchSize = logQueue.size();
+        for (int index = 0; index < batchSize; index++) {
+            entry = logQueue.poll();
+            if (entry != null) {
+                try {
+                    prepareAuditFile();
+                    entry.send(xmlSerializer);
+                    xmlSerializer.flush();
+                }
+                catch (IOException e) {
+                    Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+                    // Add again the entry to the queue to save it later
+                    logQueue.add(entry);
+                }
+                catch (XMLStreamException e) {
+                    Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+                    // Add again the entry to the queue to save it later
+                    logQueue.add(entry);
+                }
+            }
+        }
+    }
+
+    /**
+     * Wrapper on a Packet with information about the packet's status at the moment when the message
+     * was queued.<p>
+     *
+     * The idea is to wrap every packet that is needed to be audited and then add the wrapper to a
+     * queue that will be later processed (i.e. saved to the XML file).
+     */
+    private class AuditPacket {
+        private XMPPPacket packet;
+        private String streamID;
+        private String sessionStatus;
+        private Date timestamp;
+        private boolean sending;
+        private boolean dropped;
+
+        public AuditPacket(XMPPPacket packet, boolean dropped) {
+            this.packet = packet;
+            this.dropped = dropped;
+            this.timestamp = new Date();
+            this.sending = packet.isSending();
+            Session session = packet.getOriginatingSession();
+            if (session != null) {
+                if (session.getStreamID() != null) {
+                    this.streamID = session.getStreamID().toString();
+                }
+                switch (session.getStatus()) {
+                    case Session.STATUS_AUTHENTICATED:
+                        this.sessionStatus =  "auth";
+                        break;
+                    case Session.STATUS_CLOSED:
+                        this.sessionStatus = "closed";
+                        break;
+                    case Session.STATUS_CONNECTED:
+                        this.sessionStatus = "connected";
+                        break;
+                    case Session.STATUS_STREAMING:
+                        this.sessionStatus = "stream";
+                        break;
+                    default:
+                        this.sessionStatus = "unknown";
+                        break;
+                }
+            }
+        }
+
+        public void send(XMLStreamWriter xmlSerializer) {
+            try {
+                xmlSerializer.writeStartElement("packet");
+                xmlSerializer.writeDefaultNamespace("http://jivesoftware.org");
+
+                if (streamID != null) {
+                    xmlSerializer.writeAttribute("session", streamID);
+                }
+                if (sessionStatus != null) {
+                    xmlSerializer.writeAttribute("status", sessionStatus);
+                }
+                xmlSerializer.writeAttribute("timestamp", timestamp.toString());
+                if (sending) {
+                    xmlSerializer.writeAttribute("sending", "true");
+                }
+                if (dropped) {
+                    xmlSerializer.writeAttribute("dropped", "true");
+                }
+                packet.send(xmlSerializer, 0);
+                xmlSerializer.writeEndElement();
+            }
+            catch (Exception e) {
+                Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+            }
+        }
+
+    }
+
 }

src/web/audit-policy.jsp

@@ -44,6 +44,7 @@
     String[] xpathQuery = ParamUtils.getParameters(request,"xpathQuery");
     String maxCount = ParamUtils.getParameter(request,"maxCount");
     String maxSize = ParamUtils.getParameter(request,"maxSize");
+    String logTimeout = ParamUtils.getParameter(request,"logTimeout");
 
     // Get an audit manager:
     AuditManager auditManager = (AuditManager)admin.getServiceLookup().lookup(AuditManager.class);
@@ -73,6 +74,11 @@
         } catch (Exception e){
             errors.put("maxSize","maxSize");
         }
+        try {
+            auditManager.setLogTimeout(Integer.parseInt(logTimeout) * 1000);
+        } catch (Exception e){
+            errors.put("logTimeout","logTimeout");
+        }
         // All done, redirect
         if (errors.size() == 0){
         %>
@@ -101,6 +107,7 @@
         auditXPath = auditManager.isAuditXPath();
         maxCount = Integer.toString(auditManager.getMaxFileCount());
         maxSize = Integer.toString(auditManager.getMaxFileSize());
+        logTimeout = Integer.toString(auditManager.getLogTimeout() / 1000);
     }
 %>
 
@@ -184,6 +191,24 @@ and IQ packets are primarily useful for tracing and troubleshooting XMPP deploym
 
                     </td>
                 </tr>
+                <tr valign="top">
+                    <td width="1%" nowrap class="c1">
+                        Flush Interval (seconds):
+                    </td>
+                    <td width="99%">
+                        <input type="text" size="15" maxlength="50" name="logTimeout"
+                         value="<%= ((logTimeout != null) ? logTimeout : "") %>">
+
+                    <%  if (errors.get("logTimeout") != null) { %>
+
+                        <span class="jive-error-text">
+                        Please enter a valid number.
+                        </span>
+
+                    <%  } %>
+
+                    </td>
+                </tr>
                 <tr valign="top">
                     <td width="1%" nowrap class="c1">
                         Packets to audit:
@@ -230,6 +255,14 @@ and IQ packets are primarily useful for tracing and troubleshooting XMPP deploym
                         </table>
                     </td>
                 </tr>
+                <tr valign="top">
+                    <td width="1%" nowrap class="c1">
+                        Queued packets:
+                    </td>
+                    <td width="99%">
+                         <%= auditManager.getAuditor().getQueuedPacketsNumber() %>
+                    </td>
+                </tr>
                 </table>
             </td>
         </tr>