receiver.java

Logica lastest SMPP API

/*
 * Copyright (c) 1996-2001
 * Logica Mobile Networks Limited
 * All rights reserved.
 *
 * This software is distributed under Logica Open Source License Version 1.0
 * ("Licence Agreement"). You shall use it and distribute only in accordance
 * with the terms of the License Agreement.
 *
 */
package com.logica.smpp;

import java.io.IOException;

import com.logica.smpp.pdu.GenericNack;
import com.logica.smpp.pdu.InvalidPDUException;
import com.logica.smpp.pdu.PDU;
import com.logica.smpp.pdu.PDUException;
import com.logica.smpp.pdu.UnknownCommandIdException;
import com.logica.smpp.util.Queue;
import com.logica.smpp.util.Unprocessed;

/**
 * <code>Receiver</code> is class used for receiving PDUs from SMSC.
 * It can be used two ways: it has methods for synchronous (blocking)
 * receiving of PDUs and as it is derived from <code>ReceiverBase</code>
 * whic on turn is derived from <code>ProcessingThread</code> class,
 * it can also receive PDUs on background and puts them into a queue.
 * 
 * @author Logica Mobile Networks SMPP Open Source Team
 * @version 1.2, 1 Oct 2001
 * @see ReceiverBase
 * @see Connection
 * @see Session
 * @see Queue
 */

/*
  13-07-01 [e-mail] start(), stop(), setQueueWaitTimeout() &
                           getQueueWaitTimeout() made not synchronized;
                           receive(long) & receive(PDU) made synchronized
                           so the receiver no longer locks up
  13-07-01 [e-mail] bug fixed in tryReceivePDU which caused that the PDUs
                           were never removed from the queue - now dequeue(expected)
                           is now used instead of find(expected)
  13-07-01 [e-mail] loads of debug lines corrected; some added
  08-08-01 [e-mail] added support for Session's asynchronous processing capability
  26-09-01 [e-mail] debug code categorized to groups
  01-10-01 [e-mail] added function getThreadName for ProcessingThread
                           thread name initialisation.
  02-10-01 [e-mail] instead of importing full packages only the used
                           classes are iported
*/
public class Receiver extends ReceiverBase
{
    /**
     * Name of the thread created when starting
     * the <code>ProcessingThread</code>.
     * @see com.logica.smpp.util.ProcessingThread#start()
     * @see com.logica.smpp.util.ProcessingThread#generateIndexedThreadName()
     */
    private static final String RECEIVER_THREAD_NAME = "Receiver";

    /**
     * The correspondent transmitter for transmitting PDUs.
     * It's used for sending of generic negative acknowledges, if necessary.
     * It is passed to the receiver as a parameter during construction.
     * 
     * @see #receiveAsync()
     */
    private Transmitter transmitter = null;
    
    /**
     * The network connection which is used for receiving data.
     * It is passed to the receiver as a parameter during construction.
     */
    private Connection connection = null;
    
    /**
     * The queue which holds the received PDUs.
     * As the PDUs are received in asynchronnous manner, they are stored
     * to a queue from which they can be get using method
     * <code>receive</code>. PDUs are stored to the queue if and only if the
     * <code>Receiver</code> is started as a separate thread using method
     * <code>start</code>.
     *
     * @see #receive(long)
     * @see #receive(PDU)
     * @see #start()
     * @see ReceiverBase#start()
     */
    private Queue pduQueue = new Queue();
    
    /**
     * This timeout specifies for how long will go the receiving into wait
     * if the PDU (expected or any) isn't in the <code>pduQueue</code> yet.
     * After that the queue is probed again (etc.) until receiving timeout 
     * expires or the PDU is received.
     *
     * @see #tryReceivePDU(Connection,PDU)
     */
    private long queueWaitTimeout = Data.QUEUE_TIMEOUT;

    /**
     * Indication if the <code>Receiver</code> is receiving on background as an
     * extra thread.
     *
     * @see #start()
     * @see #tryReceivePDU(Connection,PDU)
     */
    private boolean receiver = false;
    
    /**
     * This object holds data received from connection which aren't complete
     * PDU yet. If this situation occurs, it's likely that the data
     * will be received the next time when another attempt
     * to receive data from the connection occurs. Its used in
     * <code>ReceiverBase</code>'s <code>receivePDUFromConnection</code>
     * method.
     *
     * @see ReceiverBase#receivePDUFromConnection(Connection,Unprocessed)
     */
    private Unprocessed unprocessed = new Unprocessed();
    
    /**
     * If the receiving is asynchronous, <code>pduListener</code> must
     * contain the callback object used for processing of PDUs received
     * from the SMSC. <code>Receiver</code> after receiving a PDU passes
     * the received PDU to apropriate member function of the processor.
     * @see #asynchronous
     * @see #setServerPDUEventListener(ServerPDUEventListener)
     */
    private ServerPDUEventListener pduListener = null;

    /**
     * Indicates that the sending of PDUs to the SMSC is asynchronous, i.e.
     * the session doesn't wait for a response to the sent request as well as
     * the <code>receive</code> functions will return null as all received 
     * PDUs are passed to the <code>pduListener</code> object in
     * the <code>receiver</code>
     * @see #pduListener
     * @see #setServerPDUEventListener(ServerPDUEventListener)
     */
    private boolean asynchronous = false;

    /**
     * This constructor sets the connection to receive the messages from.
     *
     * @param connection the connection to use for receiving
     * @see Connection
     */
    public Receiver(Connection connection)
    {
        this.connection = connection;
    }
    
    /**
     * This constructor sets the connection to receive the messages from
     * and a transmitter for sending generic negative acknowledges
     * if necessary.
     *
     * @param transmitter the transmitter to use for sending
     *                    <code>GenericNack</code>
     * @param connection  the connection to use for receiving and transmitting
     * @see Transmitter
     * @see Connection
     * @see GenericNack
     */
    public Receiver(Transmitter transmitter, Connection connection)
    {
        this.transmitter = transmitter;
        this.connection = connection;
    }
    
    /**
     * Returns if the receiver receives PDUs on background as an extra thread.
     *
     * @see #receiver
     */
    public boolean isReceiver() { return receiver; }

    /**
     * Sets the event listener for asynchronous <code>Receiver</code>.
     * In case there are unprocessed PDUs in the queue,
     * they are removed from the queue and passed to the newly set listener.
     */
    public synchronized void setServerPDUEventListener(ServerPDUEventListener pduListener)
    {
        this.pduListener = pduListener;
        this.asynchronous = pduListener != null;
        if (asynchronous) {
            // let's remove all pdu's from the queue as since now all
            // processing should be asynchronous -- it's not wise to
            // expect that the programmer will try AFTER setting the listener
            // to call receive() which when in sync mode removes the pdus from
            // the queue
            PDU pdu;
            int queueSize;
            synchronized (pduQueue) {
                queueSize = pduQueue.size();
                for (int i = 0; i < queueSize; i++) {
                    pdu = (PDU)pduQueue.dequeue();
                    process(pdu);
                }
            }
        }
    }

    /**
     * Resets unprocessed data and starts receiving on the background.
     *
     * @see ReceiverBase#start()
     */
    public void start()
    {
        debug.write(DRXTX,"Receiver starting");
        receiver = true;
        unprocessed.reset();
        super.start();
        debug.write(DRXTX,"Receiver started");
    }

    /**
     * Stops receiving on the background.
     *
     * @see ReceiverBase#stop()
     */
    public void stop()
    {
        debug.write(DRXTX,"Receiver stoping");
        if (isReceiver()) {
            super.stop();
            receiver = false;
        }
        debug.write(DRXTX,"Receiver stoped");
    }

    /**
     * This method receives a PDU or returns PDU received on background,
     * if there is any. It tries to receive a PDU for the specified timeout.
     * If the receiver is asynchronous, then no attempt to receive a PDU
     * and <code>null</code> is returned.
     * The function calls are nested as follows:<br>
     * <ul>
     *   <li>No background receiver thread<br><code>
     *       Receiver.receive(long)<br>
     *       ReceiverBase.tryReceivePDUWithTimeout(Connection,PDU,long)<br>
     *       Receiver.tryReceivePDU(Connection,PDU)<br>
     *       ReceiverBase.receivePDUFromConnection<br>
     *       Connection.receive()</code>
     *   <li>Has background receiver thread<br><code>
     *       Receiver.receive(long)<br>
     *       ReceiverBase.tryReceivePDUWithTimeout(Connection,PDU,long)<br>
     *       Receiver.tryReceivePDU(Connection,PDU)<br>
     *       Queue.dequeue(PDU)</code><br>
     *       and the ReceiverBase.run() function which actually receives the
     *       PDUs and stores them to a queue looks as follows:<br><code>