notifier.java

这是linux下ssl vpn的实现程序

/*
 *  SSL-Explorer
 *
 *  Copyright (C) 2003-2006 3SP LTD. All Rights Reserved
 *
 *  This program is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU General Public License
 *  as published by the Free Software Foundation; either version 2 of
 *  the License, or (at your option) any later version.
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public
 *  License along with this program; if not, write to the Free Software
 *  Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */
			
package com.sslexplorer.notification;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.sslexplorer.core.CoreAttributeConstants;
import com.sslexplorer.core.CoreEvent;
import com.sslexplorer.core.CoreEventConstants;
import com.sslexplorer.core.CoreServlet;
import com.sslexplorer.policyframework.Policy;
import com.sslexplorer.policyframework.Principal;
import com.sslexplorer.security.Role;
import com.sslexplorer.security.User;

public class Notifier {
    final static Log log = LogFactory.getLog(Notifier.class);
    private List messages;
    private boolean started;
    private File queueDirectory;
    private long messageId;
    private MessageConsumer messageConsumer;
    private boolean stop;
    private List sinks;
    private HashMap sinkEnabled;

    public Notifier(File queueDirectory) throws IOException {
        messages = new ArrayList();
        this.queueDirectory = queueDirectory;
        sinks = new ArrayList();
        sinkEnabled = new HashMap();
        loadFromDisk();
    }

    public List getSinks() {
        return sinks;
    }

    public void addSink(MessageSink sink, boolean enabled) {
        if(enabled && started) {
        
            try {
            sink.start(this);
            } catch(Exception ex) {
                log.error("Failed to start " + sink.getName() + " message sink", ex);
                return;
            }
        }
        
        sinks.add(sink);
        sinkEnabled.put(sink.getName(), Boolean.valueOf(enabled));
        
    }

    public void removeSink(MessageSink sink) {
        sinks.remove(sink);
        sinkEnabled.remove(sink.getName());
    }

    public boolean isEnabled(String sinkName) {
        return Boolean.TRUE.equals(sinkEnabled.get(sinkName));
    }

    public void start() throws IllegalStateException {
        if (started) {
            throw new IllegalStateException("Already started.");
        }
        messageConsumer = new MessageConsumer();
        if (sinks.size() == 0) {
            throw new IllegalStateException(
                            "At least one message sink must have been registered for the notfication queue to be started.");
        }
        for (Iterator i = sinks.iterator(); i.hasNext();) {
            MessageSink sink = (MessageSink) i.next();
            try {
            	if (log.isDebugEnabled())
            		log.debug("Starting message sink " + sink.getName());
                sink.start(this);
            } catch (Exception e) {
                log.error("Failed to start sink " + sink.getName() + ".", e);
            }
        }
        started = true;
        messageConsumer.start();
        if (log.isInfoEnabled())
        	log.info("Notifier started");
    }

    void loadFromDisk() throws IOException {
        File[] f = queueDirectory.listFiles(new FileFilter() {
            public boolean accept(File f) {
                return f.getName().endsWith(".msg");
            }
        });
        // TODO better error handling in parsing of message files. Report on
        // non-existant / unreadable directory
        if (f == null) {
            throw new IOException("Could not list queue directory " + queueDirectory.getAbsolutePath());
        }
        for (int i = 0; i < f.length; i++) {
            FileInputStream fin = new FileInputStream(f[i]);
            try {
                DataInputStream din = new DataInputStream(fin);
                long id = din.readLong();
                String sinkName = din.readUTF();
                messageId = Math.max(id, messageId);
                boolean urgent = din.readBoolean();
                String subject = din.readUTF();
                List recipientList = new ArrayList();
                while (true) {
                    int recipientType = din.readInt();
                    if (recipientType == Recipient.EOF) {
                        break;
                    } else {
                        String recipientAlias = din.readUTF();
                        Recipient recipient = new Recipient(recipientType, recipientAlias);
                        recipientList.add(recipient);
                    }
                }
                Properties parameters = new Properties();
                while (true) {
                    int parameterType = din.readInt();
                    if (parameterType < 1) {
                        break;
                    } else {
                        String key = din.readUTF();
                        String val = din.readUTF();
                        parameters.setProperty(key, val);
                    }
                }
                String content = din.readUTF();
                String lastMessage = din.readUTF();
                Message msg = new Message(subject, content, urgent);
                msg.setId(id);
                msg.setRecipients(recipientList);
                msg.setSinkName(sinkName);
                msg.setLastMessage(lastMessage);
                queue(msg);
            } finally {
                fin.close();
            }
        }
    }

    public void stop() throws IllegalStateException {
        if (!started) {
            throw new IllegalStateException("Notifier is not started.");
        }
        stop = true;
        queueNotify();
        started = false;
        if (log.isInfoEnabled())
        	log.info("Waiting for message consumer to stop");
        try {
            messageConsumer.join();
        } catch (InterruptedException ie) {
        }
        for (Iterator i = sinks.iterator(); i.hasNext();) {
            MessageSink sink = (MessageSink) i.next();
            try {
                sink.stop();
            } catch (Exception e) {
                log.error("Failed to stop sink " + sink.getName() + ".", e);
            }
        }
        if (log.isInfoEnabled())
        	log.info("Notifier stopped");
    }

    public void clearAllMessages() {
        synchronized (messages) {
            messages.clear();
            File[] f = queueDirectory.listFiles(new FileFilter() {
                public boolean accept(File f) {
                    return f.getName().endsWith(".msg");
                }
            });
            if (f != null) {
                for (int i = 0; i < f.length; i++) {
                    f[i].delete();
                }
            }
        }
    }

    public void sendToAll(Message message) {
        sendToSink("*", message);
    }

    public void sendToAllExcept(Message message, String except) {
        sendToSink("!" + except, message);
    }

    public void sendToFirst(Message message) {
        sendToSink("^", message);
    }

    public void sendToSink(String sinkName, Message message) throws IllegalArgumentException {
        messageId++;
        message.setSinkName(sinkName);
        message.setId(messageId);
        if (log.isDebugEnabled())
        	log.debug("Sending message " + message.getId() + " '" + message.getSubject() + "' to sink '" + sinkName + "'");
        queue(message);
        try {
            write(message);
        } catch (IOException ioe) {
        }
    }

    public void setEnabled(String sinkName, boolean enabled) {
        sinkEnabled.put(sinkName, Boolean.valueOf(enabled));
        if (enabled) {
            queueNotify();
        }
    }

    public List getMessages() {
        return messages;
    }

    public List getFullRecipientListAsUsers(List recipients) throws Exception {
        Map h = new TreeMap();
        for (Iterator i = recipients.iterator(); i.hasNext();) {
            Recipient r = (Recipient) i.next();
            if (r.getRecipientType() == Recipient.USER) {
                h.put(r.getRecipientAlias(), r);
            } else if (r.getRecipientType() == Recipient.POLICY) {
                User[] users = CoreServlet.getServlet().getUserDatabase().listAllUsers("*");
                Policy pol = CoreServlet.getServlet().getPolicyDatabase().getPolicy(Integer.parseInt(r.getRecipientAlias()));
                List principals = CoreServlet.getServlet().getPolicyDatabase().getPrincipalsGrantedPolicy(pol);
                for (Iterator j = principals.iterator(); j.hasNext();) {
                    Principal p = (Principal) j.next();
                    if (p instanceof Role) {
                        for (int x = 0; x < users.length; x++) {
                            Role[] roles = users[x].getRoles();
                            for (int y = 0; y < roles.length; y++) {
                                if (roles[y].getPrincipalName().equals(p.getPrincipalName())) {
                                    h.put(users[x].getPrincipalName(), new Recipient(Recipient.USER, users[x].getPrincipalName()));
                                    break;
                                }
                            }
                        }