MailTransport.java

Go to the documentation of this file.

import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;

import javax.mail.Message;
import javax.mail.Message.RecipientType;
import javax.mail.MessagingException;
import javax.mail.Session;
import javax.mail.Transport;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

/**
 *  Encapsulates mail transport agent with FIFO queue and non-blocking send
 *  capabilities.
 *  
 *  Instances of this class may be used in two different scenarios (note 
 *  the position of MailTransport.startSending() method and the usage of 
 *  MailTransport.noMoreMessages() bellow):
 *  
 *  <div style='color:darkred;font-weight:bold'>
 *  Scenario 1: Sends one ore more messages at once.</div>
 *  
 *  Usage:
 *        - create instance using constructor MailTransport.MailTransport() 
 *          (with SMTP server as parameter)
 *        - call MailTransport.enqueueMessage()
 *        - [ call MailTransport.enqueueMessage() ]...
 *        - call MailTransport.noMoreMessages() to terminate worker thread
 *        - call MailTransport.startSending()
 *        
 *  <div style='color:darkred;font-weight:bold'>
 *  Scenario 2. Works in background as relay client to SMTP server.</div>
 *
 *  Usage:
 *        - create instance using constructor MailTransport.MailTransport() 
 *          (with SMTP server as parameters)
 *        - call MailTransport.startSending()
 *        - call MailTransport.enqueueMessage()
 *        - [ call MailTransport.enqueueMessage() ]...
 *
 *  @author Mikica B Kocic
 */  
public class MailTransport implements Runnable
{
    /**
     *  Provides message presentation and call-back context for the instance
     *  of <code>MailTransport</code> class.
     */
    public interface Context
    {
        /**
         *  Logs message terminated with new line
         *  
         *  @param str    message that will be logged
         */
        public abstract void println( String str );
        
        /**
         *  On send completed call-back
         *  
         *  @param errorMessage    error message if <> null, otherwise success
         */
        public abstract void onSendCompleted( String errorMessage );
    }

    /**
     *  Instance of the main thread.
     */
    private Thread workerThread = null;

    /**
     *  Presentation contexts
     */
    private Context context = null;
    
    /**
     *  Session to SMTP server
     */
    private Session session = null;

    /**
     *  FIFO message queue served by worker thread
     */
    private LinkedBlockingQueue<Message> queue = null;

    /**
     *  The special message used to signal the worker thread to stop
     *  processing of messages and terminate.
     */
    private Message quitSignal = null;

    /**
     *  Creates instance of the mail transport with the (new) session
     *  to SMTP server  
     */
    public MailTransport( Context context, String smtpServer, int smtpPort )
    {
        this.context    = context;
        this.queue      = new LinkedBlockingQueue<Message> ();
        this.quitSignal = new MimeMessage( this.session );

        Properties relayHost = new Properties();
        relayHost.put( "mail.smtp.host", smtpServer );
        relayHost.put( "mail.smtp.port", smtpPort );

        /* Create instance of the session.
         * DO NOT USE: getDefaultInstance() as it will create cached session.
         */
        this.session = Session.getInstance( relayHost, null ); 
    }

    /**
     *  Gets number of messages in the queue
     */
    public int getQueueSize ()
    {
        return this.queue.size ();
    }

    /**
     *  Starts the worker thread (if not already started)
     */
    public void startSending ()
    {
        synchronized( this )
        {
            if ( workerThread != null ) {
                return; // already started (there can be only one worker thread)
            }

            this.workerThread = new Thread( this );
            this.workerThread.start ();
        }
    }

    /**
     *  Sends QUIT message to the worker thread
     */
    public void noMoreMessages ()
    {
        synchronized( this )
        {
            /* put the 'QUIT' message into queue
             */
            try {
                this.queue.put( this.quitSignal );
            } catch( InterruptedException e ) {
                /* ignore */
            }
        }
    }

    /**
     *  Creates and puts new MIME text message into outgoing queue
     */
    public void enqueueMessage( String from, 
            String[] to, String subject, String textMessage 
            ) throws MessagingException
    {
        InternetAddress[] toAddr = new InternetAddress[ to.length ];
        for( int i = 0; i < to.length; i++ ) {
            toAddr[i] = new InternetAddress( to[i] );
        }

        MimeMessage mimeMsg = new MimeMessage( session );
        mimeMsg.setFrom( new InternetAddress( from ) );
        mimeMsg.setRecipients( Message.RecipientType.TO, toAddr );
        mimeMsg.setHeader( "X-Mailer", "IP1-MailSender" );
        mimeMsg.setSubject( subject );
        mimeMsg.setContent( textMessage, "text/plain" );

        /* Places the message into outgoing queue...
         */
        try {
            this.queue.put( mimeMsg );
        } catch( InterruptedException e ) {
            /* ignore */
        }
    }

    /**
     *  Waits for messages to become available in the queue, then takes
     *  messages (first-in first-out) from the queue and relays them through the
     *  session to remote SMTP server. After receiving special 'QUIT' message, 
     *  the worker thread terminates. In case of error, thread also quits and 
     *  reports the error using Context.onSendCompleted() from the MailTransport.Context 
     *  call-back interface.
     */
    @Override
    public void run ()
    {
        context.println( "MailTransport(" + workerThread.getId () 
                + "): Worker thread started..." );
        
        context.println( "MailTransport(" + workerThread.getId () 
                + "): Found " + queue.size() + " messages in queue." );
        
        String errorMessage = null; // error to be reported to the owner
        int curMessageNo = 0; // keeps track of the current message id

        while( true )
        {
            Message msgToSend = null;

            /* Retrieves and removes the head message from the queue,
             * waiting if necessary until an element becomes available).
             * Terminates the thread in case of interrupt.
             */
            try {
                context.println( "MailTransport(" + workerThread.getId () 
                        + "): Waiting for messsages..." );
                msgToSend = this.queue.take ();
            } catch( InterruptedException e ) {
                break;
            }

            /* Terminates the thread if the last retrieved message is
             * the 'QUIT thread' message.
             */
            if ( msgToSend == quitSignal || msgToSend == null ) {
                context.println( "MailTransport(" + workerThread.getId () 
                        + "): Got signal to 'QUIT'..." );
                break;
            }

            /* Sends message via transport protocol
             */
            try
            {
                ++curMessageNo;
                
                context.println( "MailTransport(" + workerThread.getId () 
                        + "): Sending message #" + curMessageNo + "..." );
                
                for( javax.mail.Address a : msgToSend.getFrom() ) {
                    context.println( "    From:    " + a.toString () );
                }
                for( javax.mail.Address a : msgToSend.getRecipients( RecipientType.TO ) ) {
                    context.println( "    To:      " + a.toString () );
                }
                context.println( "    Subject: " + msgToSend.getSubject () );

                Transport.send( msgToSend );
                
                context.println( "MailTransport(" + workerThread.getId () 
                        + "): MESSAGE SENT!" );
            }
            catch( MessagingException e )
            {
                context.println( "MailTransport(" + workerThread.getId () 
                        + "): Error\n\n" + e.toString () + "\n" );
                
                errorMessage = e.toString ();

                break; // there is no point to continue in case of error...
            }
        }
        
        context.println( "MailTransport(" + workerThread.getId () 
                + "): Thread done." );
        
        context.onSendCompleted( errorMessage );
    }
}