Added interceptors support. JM-218

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

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

@@ -11,13 +11,19 @@
 
 package org.jivesoftware.messenger.audit.spi;
 
-import org.jivesoftware.messenger.container.BasicModule;
-import org.jivesoftware.messenger.audit.AuditManager;
-import org.jivesoftware.messenger.audit.Auditor;
 import org.jivesoftware.messenger.JiveGlobals;
+import org.jivesoftware.messenger.Session;
 import org.jivesoftware.messenger.XMPPServer;
+import org.jivesoftware.messenger.audit.AuditManager;
+import org.jivesoftware.messenger.audit.Auditor;
+import org.jivesoftware.messenger.container.BasicModule;
+import org.jivesoftware.messenger.interceptor.InterceptorManager;
+import org.jivesoftware.messenger.interceptor.PacketInterceptor;
+import org.xmpp.packet.Packet;
 
-import java.util.*;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
 
 public class AuditManagerImpl extends BasicModule implements AuditManager {
 
@@ -34,6 +40,7 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
     private static final int MAX_FILE_SIZE = 10;
     private static final int MAX_FILE_COUNT = 10;
     private static final int DEFAULT_LOG_TIMEOUT = 120000;
+    private AuditorInterceptor interceptor;
 
     public AuditManagerImpl() {
         super("Audit Manager");
@@ -46,6 +53,13 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
     public void setEnabled(boolean enabled) {
         this.enabled = enabled;
         JiveGlobals.setProperty("xmpp.audit.active", enabled ? "true" : "false");
+        // Add or remove the auditor interceptor depending on the enabled status
+        if (enabled) {
+            InterceptorManager.getInstance().addInterceptor(interceptor);
+        }
+        else {
+            InterceptorManager.getInstance().removeInterceptor(interceptor);
+        }
     }
 
     public Auditor getAuditor() {
@@ -163,6 +177,11 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
         auditor = new AuditorImpl(this);
         auditor.setMaxValues(maxSize, maxCount);
         auditor.setLogTimeout(logTimeout);
+
+        interceptor = new AuditorInterceptor();
+        if (enabled) {
+            InterceptorManager.getInstance().addInterceptor(interceptor);
+        }
     }
 
     public void stop() {
@@ -170,4 +189,13 @@ public class AuditManagerImpl extends BasicModule implements AuditManager {
             auditor.stop();
         }
     }
+
+    private class AuditorInterceptor implements PacketInterceptor {
+
+        public void interceptPacket(Packet packet, Session session, boolean read, boolean processed) {
+            if (!processed) {
+                auditor.audit(packet, session);
+            }
+        }
+    }
 }

src/java/org/jivesoftware/messenger/net/SocketConnection.java

@@ -11,23 +11,25 @@
 
 package org.jivesoftware.messenger.net;
 
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.io.Writer;
-import java.net.InetAddress;
-import java.net.Socket;
 import org.dom4j.io.XMLWriter;
 import org.jivesoftware.messenger.PacketDeliverer;
 import org.jivesoftware.messenger.PacketException;
 import org.jivesoftware.messenger.Session;
-import org.jivesoftware.messenger.audit.Auditor;
 import org.jivesoftware.messenger.auth.UnauthorizedException;
+import org.jivesoftware.messenger.interceptor.InterceptorManager;
+import org.jivesoftware.messenger.interceptor.PacketRejectedException;
 import org.jivesoftware.messenger.spi.BasicConnection;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.Log;
 import org.xmpp.packet.Packet;
 
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.net.InetAddress;
+import java.net.Socket;
+
 /**
  * An object to track the state of a Jabber client-server session.
  * Currently this class contains the socket channel connecting the
@@ -57,10 +59,6 @@ public class SocketConnection extends BasicConnection {
      */
     private PacketDeliverer deliverer;
 
-    /**
-     * Audits packets
-     */
-    private Auditor auditor;
     private Session session;
     private boolean secure;
     private XMLWriter xmlSerializer;
@@ -70,20 +68,18 @@ public class SocketConnection extends BasicConnection {
      * Create a new session using the supplied socket.
      *
      * @param deliverer The packet deliverer this connection will use
-     * @param auditor   Auditor that will audit outgoing packets
      * @param socket    The socket to represent
      * @param isSecure  True if this is a secure connection
      * @throws NullPointerException If the socket is null
      */
-    public SocketConnection(PacketDeliverer deliverer, Auditor auditor, Socket socket,
-            boolean isSecure) throws IOException
+    public SocketConnection(PacketDeliverer deliverer, Socket socket, boolean isSecure)
+            throws IOException
     {
         if (socket == null) {
             throw new NullPointerException("Socket channel must be non-null");
         }
 
         this.secure = isSecure;
-        this.auditor = auditor;
         sock = socket;
         writer = new BufferedWriter(new OutputStreamWriter(sock.getOutputStream(), charset));
         this.deliverer = deliverer;
@@ -180,21 +176,29 @@ public class SocketConnection extends BasicConnection {
             deliverer.deliver(packet);
         }
         else {
-            synchronized (writer) {
-                try {
-                    xmlSerializer.write(packet.getElement());
-                    if (flashClient) {
-                        writer.write('\0');
+            try {
+                // Invoke the interceptors before we send the packet
+                InterceptorManager.getInstance().invokeInterceptors(packet, session, false, false);
+                synchronized (writer) {
+                    try {
+                        xmlSerializer.write(packet.getElement());
+                        if (flashClient) {
+                            writer.write('\0');
+                        }
+                        xmlSerializer.flush();
+                    }
+                    catch (IOException e) {
+                        Log.error(e);
+                        close();
                     }
-                    xmlSerializer.flush();
-                }
-                catch (IOException e) {
-                    Log.error(e);
-                    close();
                 }
+                // Invoke the interceptors after we have sent the packet
+                InterceptorManager.getInstance().invokeInterceptors(packet, session, false, true);
+                session.incrementServerPacketCount();
+            }
+            catch (PacketRejectedException e) {
+                // An interceptor rejected the packet so do nothing
             }
-            auditor.audit(packet, session);
-            session.incrementServerPacketCount();
         }
     }
 

src/java/org/jivesoftware/messenger/net/SocketReadThread.java

@@ -14,8 +14,9 @@ package org.jivesoftware.messenger.net;
 import org.dom4j.Element;
 import org.dom4j.io.XPPPacketReader;
 import org.jivesoftware.messenger.*;
-import org.jivesoftware.messenger.audit.Auditor;
 import org.jivesoftware.messenger.auth.UnauthorizedException;
+import org.jivesoftware.messenger.interceptor.InterceptorManager;
+import org.jivesoftware.messenger.interceptor.PacketRejectedException;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.Log;
 import org.xmlpull.v1.XmlPullParser;
@@ -54,10 +55,6 @@ public class SocketReadThread extends Thread {
      * Router used to route incoming packets to the correct channels.
      */
     private PacketRouter router;
-    /**
-     * Audits incoming data
-     */
-    private Auditor auditor;
     private boolean clearSignout = false;
     XPPPacketReader reader = null;
 
@@ -76,16 +73,13 @@ public class SocketReadThread extends Thread {
      *
      * @param router     The router for sending packets that were read
      * @param serverName The name of the server this socket is working for
-     * @param auditor    The audit manager that will audit incoming packets
      * @param sock       The socket to read from
      * @param conn       The connection being read
      */
-    public SocketReadThread(PacketRouter router, String serverName, Auditor auditor, Socket sock,
-                            Connection conn) {
+    public SocketReadThread(PacketRouter router, String serverName, Socket sock, Connection conn) {
         super("SRT reader");
         this.serverName = serverName;
         this.router = router;
-        this.auditor = auditor;
         this.connection = conn;
         this.sock = sock;
     }
@@ -203,9 +197,25 @@ public class SocketReadThread extends Thread {
                     continue;
                 }
                 packet.setFrom(session.getAddress());
-                auditor.audit(packet, session);
-                router.route(packet);
-                session.incrementClientPacketCount();
+                try {
+                    // Invoke the interceptors before we process the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            false);
+                    router.route(packet);
+                    // Invoke the interceptors after we have processed the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            true);
+                    session.incrementClientPacketCount();
+                }
+                catch (PacketRejectedException e) {
+                    // An interceptor rejected this packet so answer a not_allowed error
+                    Message reply = new Message();
+                    reply.setID(packet.getID());
+                    reply.setTo(session.getAddress());
+                    reply.setFrom(packet.getTo());
+                    reply.setError(PacketError.Condition.not_allowed);
+                    session.process(reply);
+                }
             }
             else if ("presence".equals(tag)) {
                 Presence packet = null;
@@ -223,11 +233,27 @@ public class SocketReadThread extends Thread {
                     continue;
                 }
                 packet.setFrom(session.getAddress());
-                auditor.audit(packet, session);
-                router.route(packet);
-                session.incrementClientPacketCount();
-                // Update the flag that indicates if the user made a clean sign out
-                clearSignout = (Presence.Type.unavailable == packet.getType() ? true : false);
+                try {
+                    // Invoke the interceptors before we process the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            false);
+                    router.route(packet);
+                    // Invoke the interceptors after we have processed the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            true);
+                    session.incrementClientPacketCount();
+                    // Update the flag that indicates if the user made a clean sign out
+                    clearSignout = (Presence.Type.unavailable == packet.getType() ? true : false);
+                }
+                catch (PacketRejectedException e) {
+                    // An interceptor rejected this packet so answer a not_allowed error
+                    Presence reply = new Presence();
+                    reply.setID(packet.getID());
+                    reply.setTo(session.getAddress());
+                    reply.setFrom(packet.getTo());
+                    reply.setError(PacketError.Condition.not_allowed);
+                    session.process(reply);
+                }
             }
             else if ("iq".equals(tag)) {
                 IQ packet = null;
@@ -250,9 +276,26 @@ public class SocketReadThread extends Thread {
                     continue;
                 }
                 packet.setFrom(session.getAddress());
-                auditor.audit(packet, session);
-                router.route(packet);
-                session.incrementClientPacketCount();
+                try {
+                    // Invoke the interceptors before we process the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            false);
+                    router.route(packet);
+                    // Invoke the interceptors after we have processed the read packet
+                    InterceptorManager.getInstance().invokeInterceptors(packet, session, true,
+                            true);
+                    session.incrementClientPacketCount();
+                }
+                catch (PacketRejectedException e) {
+                    // An interceptor rejected this packet so answer a not_allowed error
+                    IQ reply = new IQ();
+                    reply.setChildElement(packet.getChildElement().createCopy());
+                    reply.setID(packet.getID());
+                    reply.setTo(session.getAddress());
+                    reply.setFrom(packet.getTo());
+                    reply.setError(PacketError.Condition.not_allowed);
+                    session.process(reply);
+                }
             }
             else {
                 throw new XmlPullParserException(LocaleUtils.getLocalizedString("admin.error.packet.tag") + tag);

src/java/org/jivesoftware/messenger/spi/ConnectionManagerImpl.java

@@ -12,7 +12,6 @@
 package org.jivesoftware.messenger.spi;
 
 import org.jivesoftware.messenger.*;
-import org.jivesoftware.messenger.audit.AuditManager;
 import org.jivesoftware.messenger.container.BasicModule;
 import org.jivesoftware.messenger.net.SSLSocketAcceptThread;
 import org.jivesoftware.messenger.net.SocketAcceptThread;
@@ -35,7 +34,6 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
     private SSLSocketAcceptThread sslSocketThread;
     private ArrayList<ServerPort> ports;
 
-    private AuditManager auditManager;
     private SessionManager sessionManager;
     private PacketDeliverer deliverer;
     private PacketRouter router;
@@ -48,9 +46,9 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
     }
 
     private void createSocket() {
-        if (!isStarted ||  isSocketStarted || auditManager == null ||
-                sessionManager == null || deliverer == null ||
-                router == null || serverName == null)
+        if (!isStarted || isSocketStarted || sessionManager == null || deliverer == null ||
+                router == null ||
+                serverName == null)
         {
             return;
         }
@@ -109,12 +107,8 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
     public void addSocket(Socket sock, boolean isSecure)  {
         try {
             // the order of these calls is critical (stupid huh?)
-            Connection conn = new SocketConnection(deliverer,
-                    auditManager.getAuditor(),
-                    sock,
-                    isSecure);
-            SocketReadThread reader = new SocketReadThread(router, serverName,
-                    auditManager.getAuditor(), sock, conn);
+            Connection conn = new SocketConnection(deliverer, sock, isSecure);
+            SocketReadThread reader = new SocketReadThread(router, serverName, sock, conn);
             reader.setDaemon(true);
             reader.start();
         }
@@ -128,7 +122,6 @@ public class ConnectionManagerImpl extends BasicModule implements ConnectionMana
         this.server = server;
         router = server.getPacketRouter();
         deliverer = server.getPacketDeliverer();
-        auditManager = server.getAuditManager();
         sessionManager = server.getSessionManager();
     }