inboxaction.java

CRM源码This file describes some issues that should be implemented in future and how it should be imple

/*
 * Copyright 2006-2007 Queplix Corp.
 *
 * Licensed under the Queplix Public License, Version 1.1.1 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.queplix.com/solutions/commercial-open-source/queplix-public-license/
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.queplix.core.modules.inbox.actions;

import com.queplix.core.error.GenericSystemException;
import com.queplix.core.modules.attachment.ejb.AttachmentManagerLocal;
import com.queplix.core.modules.attachment.ejb.AttachmentManagerLocalHome;
import com.queplix.core.modules.eql.ejb.AsyncMDB;
import com.queplix.core.modules.inbox.Account;
import com.queplix.core.modules.inbox.InboxHelper;
import com.queplix.core.modules.inbox.InboxMessage;
import com.queplix.core.modules.inbox.ejb.InboxManagerLocal;
import com.queplix.core.modules.inbox.utils.DefaultMessageFilter;
import com.queplix.core.modules.inbox.utils.InboxMsgBuilder;
import com.queplix.core.modules.inbox.utils.InboxPropertyFactory;
import com.queplix.core.modules.inbox.utils.InboxProvider;
import com.queplix.core.modules.inbox.utils.MailFilter;
import com.queplix.core.modules.inbox.utils.ResultSet;
import com.queplix.core.modules.inbox.utils.log.SystemLogPublisher;
import com.queplix.core.modules.mail.Attachment;
import com.queplix.core.integrator.security.LogonSession;
import com.queplix.core.modules.services.XAAction;
import com.queplix.core.modules.services.XAActionContext;
import com.queplix.core.utils.DateHelper;
import com.queplix.core.utils.JNDINames;
import com.queplix.core.utils.StringHelper;
import com.queplix.core.utils.async.ASyncObject;
import com.queplix.core.utils.async.ASyncRequest;
import com.queplix.core.utils.async.JMSClient;
import com.queplix.core.utils.log.AbstractLogger;
import com.queplix.core.utils.log.Log;

import javax.mail.Message;
import javax.mail.MessagingException;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

import org.apache.regexp.RE;
import org.apache.regexp.RESyntaxException;

/**
 * The service scans incomming emails.
 * @author Konstantin Mironov
 * @since 8 Dec 2006
 */
public class InboxAction extends XAAction {

    // ------------------------------------------------------- Fields

    // Service and log names.
    private String serviceName;
    private String logName;

    // Required service parameters:
    private Account account;
    private Long accountID;
    // == or ==
    private InboxMessage inboxMessage;

    // Optional service parameters.
    private Integer batchSize;
    private Boolean updateAccount;

    private LogonSession ls;
    private SystemLogPublisher publisher;

    // Async client.
    private AsyncClient asyncClient;

    // AttachmentManager EJB local interface
    private AttachmentManagerLocal attachMgr;

    private static final AbstractLogger logger = Log.getLog(InboxAction.class);

    // ------------------------------------------------------- API methods

    /*
     * No javadoc
     * @see Action#perform
     */
    public java.io.Serializable perform() {

        long time = System.currentTimeMillis();
        boolean error = false;

        // Initialization.
        init();

        try {
            if (inboxMessage != null) {
                // Performing concrete InboxMessage.
                __perform(inboxMessage);
            } else {
                // Performing all set of messages.
                __perform();
            }

        } catch (MessagingException me) {
            // Write error log.
            error = true;

            String msg = "Messaging exception: " + me.getMessage();
            publisher.ERROR(msg, accountID);
            logger.ERROR(logName + msg, me);

        } catch (Throwable t) {
            // Write error log.
            error = true;

            String msg = "Unknown exception: " + t.getMessage();
            publisher.ERROR(msg, accountID);
            logger.ERROR(logName + msg, t);

        } finally {

            // The end.
            time = (System.currentTimeMillis() - time) / 1000;
            if (error) {

                // If check failed, mark the account as invalid.
                if (Boolean.TRUE.equals(updateAccount)) {
                    account.setValidFlag(Boolean.FALSE);
                }

                String msg = "Service '" + serviceName + "' completed with errors. " +
                    "Time = " + time + " sec.";
                publisher.WARN(msg, accountID);
                logger.WARN(logName + msg);

            } else {

                // If checked successfully, mark the account as valid.
                if (Boolean.TRUE.equals(updateAccount)) {
                    account.setValidFlag(Boolean.TRUE);
                }

                String msg = "Service '" + serviceName + "' finished. Time = " + time + " sec.";
                publisher.INFO(msg, accountID);
                logger.INFO(logName + msg);
            } // if (error)
        } // try

        return null;

    } // perform() : java.io.Serializable

    // ------------------------------------------------------- Private methods

    /**
     * Performs set of messages.
     * @throws MessagingException the message has wrong format
     * @throws IOException saving error
     */
    private void __perform() throws MessagingException, IOException {

        // Initialization.
        int processedMessages = 0;
        Date msgMaxServerDate = null;
        boolean deleteMailFlag = Boolean.TRUE.equals(account.getDeleteMailFlag());
        boolean markEmailAsSeenFlag = Boolean.TRUE.equals(account.getMarkEmailAsSeenFlag());
        String lastMessageUID = null;
        String lastMessageDigest = null;

        // Init Inbox provider.
        Class providerClazz = InboxPropertyFactory.getInstance().getInboxProviderClass(account.getProviderName());
        InboxProvider provider = (InboxProvider)InboxPropertyFactory.getInstance().getInstance(providerClazz, account, logName, publisher);

        try {
            // Opening mail session...
            provider.open();

            // Get mail ResultSet.
            DefaultMessageFilter filter = new DefaultMessageFilter(account, provider);
            ResultSet rs = provider.getResultSet(filter);

            // Messages processing...
            Message message;
            while((message = rs.next()) != null) {
                int id = message.getMessageNumber();

                logger.DEBUG(logName + "Building message #" + id);
                logger.DEBUG(logName + "      Subject: " + message.getSubject());

                // Init Inbox message builder.
                Class msgBuilderClazz = InboxPropertyFactory.getInstance().getMessageBuilderClass();
                InboxMsgBuilder builder = (InboxMsgBuilder)InboxPropertyFactory.getInstance().
                    getInstance(msgBuilderClazz, account, logName, publisher);
                builder.setInboxProvider(provider);

                // Do build InboxMessage.
                InboxMessage im;
                try {
                    // create InboxMessage object
                    im = builder.build(message);
                } catch(MessagingException me) {
                    // just report problem
                    String msg = "Cannot parse message #" + id + " with subject '" +
                        message.getSubject() + "' due to messaging exception: " + me.getMessage();

                    publisher.ERROR(msg, accountID);
                    logger.ERROR(logName + msg, me);
                    continue;
                } // try


                // check message digest
                String lastServerMessageDigest = account.getMessageDigest();
                String messageDigest = im.getMessageDigest();
                if (lastServerMessageDigest != null && messageDigest != null) {
                    if (lastServerMessageDigest.compareTo(messageDigest) == 0) {
                        // just report problem
                        String msg = "Message #" + id + " skipped - message Digect is the same as the Last Message Digect. " +
                                "Message Digest: " + messageDigest +
                                ". Last Message Digest: " + lastServerMessageDigest;

                        publisher.INFO(msg, accountID);
                        logger.INFO(logName + msg);

                        // Set 'Max message date'.
                        Date msgServerDate = provider.getServerDate(message);
                        if (msgServerDate != null) {
                            if (msgMaxServerDate == null || msgServerDate.after(msgMaxServerDate)) {
                                msgMaxServerDate = msgServerDate;
                            }
                        } // if (msgServerDate != null)
                        continue;
                    } // if (lastServerMessageDigest.compareTo(messageDigest) == 0)
                } // if (messageUID != null && currentUID != null)

                // Process parsed message.
                try {
                    process(id, im);
                } catch(Exception ex) {
                    // just report problem
                    String msg = "Cannot process message #" + id + " with subject '" +
                        message.getSubject() + "' due to the issue: " + ex.getMessage();

                    publisher.ERROR(msg, accountID);
                    logger.ERROR(logName + msg, ex);

                    // Set 'Max message date'.
                    Date msgServerDate = provider.getServerDate(message);
                    if (msgServerDate != null) {
                        if (msgMaxServerDate == null || msgServerDate.after(msgMaxServerDate)) {
                            msgMaxServerDate = msgServerDate;
                        }
                    } // if (msgServerDate != null)
                    continue;
                } // try

                // Deleting PROCESSED messages.
                if (deleteMailFlag) {
                    DEBUG("Set delete flag for #" + id + " message.");
                    provider.setDeletedMsgFlag(message);

                } else if (markEmailAsSeenFlag) {
                    DEBUG("Set seen flag for #" + id + " message.");
                    provider.setSeenMsgFlag(message);
                } // if (deleteMailFlag)

                // Set 'Max message date'.
                Date msgServerDate = im.getReceiveTime();
                if (msgServerDate != null) {
                    if (msgMaxServerDate == null || msgServerDate.after(msgMaxServerDate)) {
                        msgMaxServerDate = msgServerDate;
                    }
                } // if (msgServerDate != null)

                // Checking the messages process-at-once limit.
                processedMessages++;
                if (batchSize != null && processedMessages >= batchSize.intValue()) {
                    break;
                }
                lastMessageUID = im.getMessageUid();
                lastMessageDigest = im.getMessageDigest();

            } // while

            String msg = "Processed " + processedMessages + " message(s).";
            if (msgMaxServerDate != null) {
                msg += " Last processed message server date: " + DateHelper.formatDate(msgMaxServerDate);
            } // if (msgMaxServerDate != null)

            publisher.INFO(msg, accountID);
            logger.INFO(logName + msg);

        } finally {
            try {
                provider.close(deleteMailFlag); // expunges all deleted messages (if deleteMailFlag = true)
                logger.INFO("Folder closed. Delete flag: " + deleteMailFlag);
            } catch(Exception ex) {
                logger.ERROR("Cannot close folder. Delete flag: " + deleteMailFlag, ex);
            } // try

            logger.INFO("Check and set last Receive Date");

            // Update account's 'Last receive date'.
            Date accLastReceiveDate = null;
            if (account.getLastReceiveDate() != null)
                accLastReceiveDate = account.getLastReceiveDate();

            logger.DEBUG("Account 'Last Received Date': " + accLastReceiveDate);
            logger.DEBUG("Processed message 'Max Server Date': " + msgMaxServerDate);

            if (msgMaxServerDate != null) {
                if (accLastReceiveDate == null || msgMaxServerDate.after(accLastReceiveDate)) {
                    account.setLastReceiveDate(msgMaxServerDate, lastMessageUID, lastMessageDigest);
                } else {
                    account.setLastMessageID(lastMessageUID, lastMessageDigest);
                } // if (accLastReceiveDate == null || msgMaxServerDate.after(accLastReceiveDate))
            } // if (msgMaxServerDate != null)
        } // try

    } // __perform()

    /**
     * Performs <code>msg</code> message.
     * @param im InboxMessage
     * @throws MessagingException the message has wrong format
     * @throws IOException saving error
     */
    private void __perform(InboxMessage im)
        throws MessagingException, IOException {

        process(1, im);

    } // __perform(InboxMessage)

    /**
     * Process one email message.