receiver.java

Short Message Peer to Peer

/*
 * 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 org.smpp;

import java.io.IOException;

import org.smpp.pdu.GenericNack;
import org.smpp.pdu.PDU;
import org.smpp.pdu.InvalidPDUException;
import org.smpp.pdu.PDUException;
import org.smpp.pdu.UnknownCommandIdException;
import org.smpp.util.Queue;
import org.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 $Revision: 1.2 $
 * @see ReceiverBase
 * @see Connection
 * @see Session
 * @see Queue
 */

public class Receiver extends ReceiverBase {
	/**
	 * Name of the thread created when starting
	 * the <code>ProcessingThread</code>.
	 * @see org.smpp.util.ProcessingThread#start()
	 * @see org.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>
	 *       ReceiverBase.run()<br>
	 *       Receiver.receiveAsync()<br>
	 *       ReceiverBase.receivePDUFromConnection<br>
	 *       Connection.receive()</code>
	 *
	 * @param timeout for how long is tried to receive a PDU
	 * @return the received PDU or null if none received for the spec. timeout
	 * 
	 * @exception IOException exception during communication
	 * @exception PDUException incorrect format of PDU
	 * @exception TimeoutException rest of PDU not received for too long time
	 * @exception UnknownCommandIdException PDU with unknown id was received
	 * @see ReceiverBase#tryReceivePDUWithTimeout(Connection,PDU,long)
	 */
	public synchronized PDU receive(long timeout)
		throws UnknownCommandIdException, TimeoutException, NotSynchronousException, PDUException, IOException {
		PDU pdu = null;
		if (!asynchronous) {