<commit_before>package org.jsmpp.session;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.SocketTimeoutException;
import java.util.Hashtable;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

import org.jsmpp.BindType;
import org.jsmpp.DefaultPDUReader;
import org.jsmpp.DefaultPDUSender;
import org.jsmpp.InterfaceVersion;
import org.jsmpp.InvalidCommandLengthException;
import org.jsmpp.InvalidResponseException;
import org.jsmpp.NumberingPlanIndicator;
import org.jsmpp.PDUReader;
import org.jsmpp.PDUSender;
import org.jsmpp.PDUStringException;
import org.jsmpp.SMPPConstant;
import org.jsmpp.SynchronizedPDUSender;
import org.jsmpp.TypeOfNumber;
import org.jsmpp.bean.BindResp;
import org.jsmpp.bean.Command;
import org.jsmpp.bean.DataCoding;
import org.jsmpp.bean.DeliverSm;
import org.jsmpp.bean.ESMClass;
import org.jsmpp.bean.EnquireLinkResp;
import org.jsmpp.bean.QuerySmResp;
import org.jsmpp.bean.RegisteredDelivery;
import org.jsmpp.bean.SubmitSmResp;
import org.jsmpp.bean.UnbindResp;
import org.jsmpp.extra.NegativeResponseException;
import org.jsmpp.extra.PendingResponse;
import org.jsmpp.extra.ProcessRequestException;
import org.jsmpp.extra.ResponseTimeoutException;
import org.jsmpp.extra.SessionState;
import org.jsmpp.session.connection.Connection;
import org.jsmpp.session.connection.ConnectionFactory;
import org.jsmpp.session.connection.socket.SocketConnectionFactory;
import org.jsmpp.session.state.SMPPSessionState;
import org.jsmpp.util.DefaultComposer;
import org.jsmpp.util.IntUtil;
import org.jsmpp.util.Sequence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author uudashr
 * @version 2.0
 *
 */
public class SMPPSession {
	private static final Logger logger = LoggerFactory.getLogger(SMPPSession.class);
	private static final Random random = new Random();
	/* Utility */
	private final PDUSender pduSender;
    private final PDUReader pduReader;
	
    /* Connection */
	private final ConnectionFactory connFactory;
	private Connection conn;
	private DataInputStream in;
	private OutputStream out;
	
	private SMPPSessionState stateProcessor = SMPPSessionState.CLOSED;
	private final Sequence sequence = new Sequence(1);
	private final ResponseHandler responseHandler = new ResponseHandlerImpl();
	private final Hashtable<Integer, PendingResponse<? extends Command>> pendingResponse = new Hashtable<Integer, PendingResponse<? extends Command>>();
	private int sessionTimer = 5000;
	private long transactionTimer = 2000;
	private long lastActivityTimestamp;
	private MessageReceiverListener messageReceiverListener;
	private SessionStateListener sessionStateListener;
    
	private EnquireLinkSender enquireLinkSender;
	private String sessionId = generateSessionId();
	
	public SMPPSession() {
        this(new SynchronizedPDUSender(new DefaultPDUSender(new DefaultComposer())), 
                new DefaultPDUReader(), 
                SocketConnectionFactory.getInstance());
    }
	
	public SMPPSession(PDUSender pduSender, PDUReader pduReader, ConnectionFactory connFactory) {
	    this.pduSender = pduSender;
	    this.pduReader = pduReader;
	    this.connFactory = connFactory;
	    
    }
	
	public SMPPSession(String host, int port, BindParameter bindParam,
            PDUSender pduSender, PDUReader pduReader,
            ConnectionFactory connFactory) throws IOException {
	    this(pduSender, pduReader, connFactory);
	    connectAndBind(host, port, bindParam);
	}
	
	public SMPPSession(String host, int port, BindParameter bindParam) throws IOException {
        this();
        connectAndBind(host, port, bindParam);
    }
	
	/**
	 * Open connection and bind immediately.
	 * 
	 * @param host is the SMSC host address.
	 * @param port is the SMSC listen port.
	 * @param bindType is the bind type.
	 * @param systemId is the system id.
	 * @param password is the password.
	 * @param systemType is the system type.
	 * @param addrTon is the address TON.
	 * @param addrNpi is the address NPI.
	 * @param addressRange is the address range.
	 * @throws IOException if there is an IO error found.
	 */
	public void connectAndBind(String host, int port, BindType bindType,
            String systemId, String password, String systemType,
            TypeOfNumber addrTon, NumberingPlanIndicator addrNpi,
            String addressRange) throws IOException {
	    connectAndBind(host, port, new BindParameter(bindType, systemId,
                password, systemType, addrTon, addrNpi, addressRange));
	}
	
	/**
	 * Open connection and bind immediately.
	 * 
	 * @param host is the SMSC host address.
	 * @param port is the SMSC listen port.
	 * @param bindParam is the bind parameters.
	 * @return the SMSC system id.
	 * @throws IOException if there is an IO error found.
	 */
	public String connectAndBind(String host, int port, BindParameter bindParam) throws IOException {
	    logger.debug("Connect and bind to " + host + " port " + port);
		if (sequence.currentValue() != 1) {
			throw new IOException("Failed connecting");
		}
		
		conn = connFactory.createConnection(host, port);
		logger.info("Connected");
		
		changeState(SessionState.OPEN);
		try {
			in = new DataInputStream(conn.getInputStream());
			out = conn.getOutputStream();
			
			new PDUReaderWorker().start();
			String smscSystemId = sendBind(bindParam.getBindType(), bindParam.getSystemId(), bindParam.getPassword(), bindParam.getSystemType(),
                    InterfaceVersion.IF_34, bindParam.getAddrTon(), bindParam.getAddrNpi(), bindParam.getAddressRange());
			changeToBoundState(bindParam.getBindType());
			
			enquireLinkSender = new EnquireLinkSender();
			enquireLinkSender.start();
			return smscSystemId;
		} catch (PDUStringException e) {
		    logger.error("Failed sending bind command", e);
		    throw new IOException("Failed sending bind since some string parameter area invalid : " + e.getMessage());
		} catch (NegativeResponseException e) {
			String message = "Receive negative bind response";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (InvalidResponseException e) {
			String message = "Receive invalid response of bind";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (ResponseTimeoutException e) {
			String message = "Waiting bind response take time to long";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (IOException e) {
			logger.error("IO Error occur", e);
			close();
			throw e;
		}
	}
	
	/**
	 * Sending bind with 1 minutes timeout.
	 *  
	 * @param bindType is the bind type.
     * @param systemId is the system id.
     * @param password is the password.
     * @param systemTypeis the system type.
     * @param interfaceVersion is the interface version.
     * @param addrTon is the address TON.
     * @param addrNpi is the address NPI.
     * @param addressRange is the address range.
     * @return SMSC system id.
     * @throws PDUStringException if we enter invalid bind parameter(s).
     * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
     * @throws InvalidResponseException if there is invalid response found.
     * @throws NegativeResponseException if we receive negative response.
     * @throws IOException if there is an IO error occur.
	 */
	private String sendBind(BindType bindType, String systemId,
            String password, String systemType,
            InterfaceVersion interfaceVersion, TypeOfNumber addrTon,
            NumberingPlanIndicator addrNpi, String addressRange)
            throws PDUStringException, ResponseTimeoutException,
            InvalidResponseException, NegativeResponseException, IOException {
	    return sendBind(bindType, systemId, password, systemType, interfaceVersion, addrTon, addrNpi, addressRange, 60000);
	}
	
	/**
	 * Sending bind.
	 * 
	 * @param bindType is the bind type.
	 * @param systemId is the system id.
	 * @param password is the password.
	 * @param systemTypeis the system type.
	 * @param interfaceVersion is the interface version.
	 * @param addrTon is the address TON.
	 * @param addrNpi is the address NPI.
	 * @param addressRange is the address range.
	 * @param timeout is the max time waiting for bind response. 
	 * @return SMSC system id.
	 * @throws PDUStringException if we enter invalid bind parameter(s).
	 * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
	 * @throws InvalidResponseException if there is invalid response found.
	 * @throws NegativeResponseException if we receive negative response.
	 * @throws IOException if there is an IO error occur.
	 */
	private String sendBind(BindType bindType, String systemId,
			String password, String systemType,
			InterfaceVersion interfaceVersion, TypeOfNumber addrTon,
			NumberingPlanIndicator addrNpi, String addressRange, long timeout)
			throws PDUStringException, ResponseTimeoutException,
			InvalidResponseException, NegativeResponseException, IOException {
		int seqNum = sequence.nextValue();
		PendingResponse<BindResp> pendingResp = new PendingResponse<BindResp>(timeout);
		pendingResponse.put(seqNum, pendingResp);
		
		try {
			pduSender.sendBind(out, bindType, seqNum, systemId, password, systemType, interfaceVersion, addrTon, addrNpi, addressRange);
		} catch (IOException e) {
			logger.error("Failed sending bind command", e);
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		try {
			pendingResp.waitDone();
			logger.info("Bind response received");
		} catch (ResponseTimeoutException e) {
			pendingResponse.remove(seqNum);
			throw e;
		} catch (InvalidResponseException e) {
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK) {
			throw new NegativeResponseException(pendingResp.getResponse().getCommandStatus());
		}
		
		return pendingResp.getResponse().getSystemId();
	}

	/**
	 * Submit short message with specified parameter.
	 * 
     * @param serviceType is the service_type parameter.
     * @param sourceAddrTon is the source_addr_ton parameter.
     * @param sourceAddrNpi is the source_addr_npi parameter.
     * @param sourceAddr is the source_addr parameter.
     * @param destAddrTon is the dest_addr_ton parameter.
     * @param destAddrNpi is the dest_addr_npi parameter.
     * @param destinationAddr is the destination_addr parameter.
     * @param esmClass is the esm_class parameter.
     * @param protocoId is the protocol_id parameter.
     * @param priorityFlag is the priority_flag parameter.
     * @param scheduleDeliveryTime is the schedule_delivery_time parameter. (It is time in string format).
     * @param validityPeriod is the validity_period parameter (It is time in string format).
     * @param registeredDelivery is the registered_delivery parameter.
     * @param replaceIfPresentFlag is the replace_if_present_flag parameter
     * @param dataCoding is the data_coding parameter.
     * @param smDefaultMsgId is the sm_default_message_id parameter.
     * @param shortMessage is the short message.
     * @return the message id from SMSC.
     * @throws PDUStringException if we enter invalid bind parameter(s).
     * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
     * @throws InvalidResponseException if there is invalid response found.
     * @throws NegativeResponseException if we receive negative response.
     * @throws IOException if there is an IO error found.
     */
    public String submitShortMessage(String serviceType,
    		TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
    		String sourceAddr, TypeOfNumber destAddrTon,
    		NumberingPlanIndicator destAddrNpi, String destinationAddr,
    		ESMClass esmClass, byte protocoId, byte priorityFlag,
    		String scheduleDeliveryTime, String validityPeriod,
    		RegisteredDelivery registeredDelivery, byte replaceIfPresentFlag,
    		DataCoding dataCoding, byte smDefaultMsgId, byte[] shortMessage)
    		throws PDUStringException, ResponseTimeoutException,
    		InvalidResponseException, NegativeResponseException, IOException {
    	
    	int seqNum = sequence.nextValue();
    	PendingResponse<SubmitSmResp> pendingResp = new PendingResponse<SubmitSmResp>(transactionTimer);
    	pendingResponse.put(seqNum, pendingResp);
    	try {
    		pduSender.sendSubmitSm(out, seqNum, serviceType, sourceAddrTon,
    				sourceAddrNpi, sourceAddr, destAddrTon, destAddrNpi,
    				destinationAddr, esmClass, protocoId, priorityFlag,
    				scheduleDeliveryTime, validityPeriod, registeredDelivery,
    				replaceIfPresentFlag, dataCoding, smDefaultMsgId, shortMessage);
    		
    	} catch (IOException e) {
    		logger.error("Failed submit short message", e);
    		pendingResponse.remove(seqNum);
    		close();
    		throw e;
    	}
    	
    	try {
    		pendingResp.waitDone();
    		logger.debug("Submit sm response received");
    	} catch (ResponseTimeoutException e) {
    		pendingResponse.remove(seqNum);
    		logger.debug("Response timeout for submit_sm with sessionIdSequence number " + seqNum);
    		throw e;
    	} catch (InvalidResponseException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	}
    	
    	if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
    		throw new NegativeResponseException(pendingResp.getResponse().getCommandStatus());
    	
    	return pendingResp.getResponse().getMessageId();
    }

    public QuerySmResult queryShortMessage(String messageId, TypeOfNumber sourceAddrTon,
    		NumberingPlanIndicator sourceAddrNpi, String sourceAddr)
    		throws PDUStringException, ResponseTimeoutException,
    		InvalidResponseException, NegativeResponseException, IOException {
    	
    	int seqNum = sequence.nextValue();
    	PendingResponse<QuerySmResp> pendingResp = new PendingResponse<QuerySmResp>(transactionTimer);
    	pendingResponse.put(seqNum, pendingResp);
    	try {
    		pduSender.sendQuerySm(out, seqNum, messageId, sourceAddrTon, sourceAddrNpi, sourceAddr);
    		
    	} catch (IOException e) {
    		logger.error("Failed submit short message", e);
    		pendingResponse.remove(seqNum);
    		close();
    		throw e;
    	}
    	
    	try {
    		pendingResp.waitDone();
    		logger.info("Query sm response received");
    	} catch (ResponseTimeoutException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	} catch (InvalidResponseException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	}
    	
    	QuerySmResp resp = pendingResp.getResponse();
    	if (resp.getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
    		throw new NegativeResponseException(resp.getCommandStatus());
    	
    	if (resp.getMessageId().equals(messageId)) {
    		return new QuerySmResult(resp.getFinalDate(), resp.getMessageState(), resp.getErrorCode());
    	} else {
    		// message id requested not same as the returned
    		throw new InvalidResponseException("Requested message_id doesn't match with the result");
    	}
    }

    public String getSessionId() {
    	return sessionId;
    }

    private void unbind() throws ResponseTimeoutException, InvalidResponseException, IOException {
		int seqNum = sequence.nextValue();
		PendingResponse<UnbindResp> pendingResp = new PendingResponse<UnbindResp>(transactionTimer);
		pendingResponse.put(seqNum, pendingResp);
		
		try {
			pduSender.sendUnbind(out, seqNum);
		} catch (IOException e) {
			logger.error("Failed sending unbind", e);
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		try {
			pendingResp.waitDone();
			logger.info("Unbind response received");
			changeState(SessionState.UNBOUND);
		} catch (ResponseTimeoutException e) {
			pendingResponse.remove(seqNum);
			throw e;
		} catch (InvalidResponseException e) {
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
			logger.warn("Receive NON-OK response of unbind");
	}

	private void changeToBoundState(BindType bindType) {
		if (bindType.equals(BindType.BIND_TX)) {
			changeState(SessionState.BOUND_TX);
		} else if (bindType.equals(BindType.BIND_RX)) {
			changeState(SessionState.BOUND_RX);
		} else if (bindType.equals(BindType.BIND_TRX)) {
			changeState(SessionState.BOUND_TRX);
		} else {
			throw new IllegalArgumentException("Bind type " + bindType + " not supported");
		}
		
		try {
			conn.setSoTimeout(sessionTimer);
		} catch (IOException e) {
			logger.error("Failed setting so_timeout for session timer", e);
		}
	}

	private synchronized void changeState(SessionState newState) {
        if (!stateProcessor.getSessionState().equals(newState)) {
            final SessionState oldState = stateProcessor.getSessionState();
    		
    		// change the session state processor
    		if (newState == SessionState.OPEN) {
    			stateProcessor = SMPPSessionState.OPEN;
    		} else if (newState == SessionState.BOUND_RX) {
    			stateProcessor = SMPPSessionState.BOUND_RX;
    		} else if (newState == SessionState.BOUND_TX) {
    			stateProcessor = SMPPSessionState.BOUND_TX;
    		} else if (newState == SessionState.BOUND_TRX) {
    			stateProcessor = SMPPSessionState.BOUND_TRX;
    		} else if (newState == SessionState.UNBOUND) {
    			stateProcessor = SMPPSessionState.UNBOUND;
    		} else if (newState == SessionState.CLOSED) {
    			stateProcessor = SMPPSessionState.CLOSED;
    		}
            fireChangeState(newState, oldState);
        }
	}
	
	public int getSessionTimer() {
		return sessionTimer;
	}
	
	public void setSessionTimer(int sessionTimer) {
		this.sessionTimer = sessionTimer;
		if (stateProcessor.getSessionState().isBound()) {
			try {
				conn.setSoTimeout(sessionTimer);
			} catch (IOException e) {
				logger.error("Failed setting so_timeout for session timer", e);
			}
		}
	}
	
	public long getTransactionTimer() {
		return transactionTimer;
	}
	
	public void setTransactionTimer(long transactionTimer) {
		this.transactionTimer = transactionTimer;
	}
	
	public synchronized SessionState getSessionState() {
		return stateProcessor.getSessionState();
	}
	
	public SessionStateListener getSessionStateListener() {
        return sessionStateListener;
    }
	
    public void setSessionStateListener(
            SessionStateListener sessionStateListener) {
        this.sessionStateListener = sessionStateListener;
    }
    
    public MessageReceiverListener getMessageReceiverListener() {
        return messageReceiverListener;
    }
    
	public void setMessageReceiverListener(
			MessageReceiverListener messageReceiverListener) {
		this.messageReceiverListener = messageReceiverListener;
	}
	
    /**
     * This method provided for monitoring need.
     * 
     * @return the last activity timestamp.
     */
    public long getLastActivityTimestamp() {
        return lastActivityTimestamp;
    }

    public void unbindAndClose() {
    	try {
    		unbind();
    	} catch (ResponseTimeoutException e) {
    		logger.error("Timeout waiting unbind response", e);
    	} catch (InvalidResponseException e) {
    		logger.error("Receive invalid unbind response", e);
    	} catch (IOException e) {
    		logger.error("IO error found ", e);
    	}
    	close();
    }

    public void close() {
		changeState(SessionState.CLOSED);
		try {
            conn.close();
        } catch (IOException e) {
        }
	}
	
	private synchronized boolean isReadPdu() {
		return stateProcessor.getSessionState().isBound() || stateProcessor.getSessionState().equals(SessionState.OPEN);
	}
	
	@Override
	protected void finalize() throws Throwable {
		close();
	}
	
	private void fireAcceptDeliverSm(DeliverSm deliverSm) throws ProcessRequestException {
		if (messageReceiverListener != null) {
			messageReceiverListener.onAcceptDeliverSm(deliverSm);
        } else { 
			logger.warn("Receive deliver_sm but MessageReceiverListener is null. Short message = " + new String(deliverSm.getShortMessage()));
        }
	}
	
    private void fireChangeState(SessionState newState, SessionState oldState) {
        if (sessionStateListener != null) {
            sessionStateListener.onStateChange(newState, oldState, this);
        } else {
            logger.warn("SessionStateListener is null");
        }
    }
    
    private synchronized static final String generateSessionId() {
        return IntUtil.toHexString(random.nextInt());
    }
    
	private class ResponseHandlerImpl implements ResponseHandler {
		
		public void processDeliverSm(DeliverSm deliverSm) throws ProcessRequestException {
			fireAcceptDeliverSm(deliverSm);
		}
		
		@SuppressWarnings("unchecked")
		public PendingResponse<Command> removeSentItem(int sequenceNumber) {
			return (PendingResponse<Command>)pendingResponse.remove(sequenceNumber);
		}
		
		public void notifyUnbonded() {
		    changeState(SessionState.UNBOUND);
		}
		
		public void sendDeliverSmResp(int sequenceNumber) throws IOException {
			try {
				pduSender.sendDeliverSmResp(out, sequenceNumber);
				logger.debug("deliver_sm_resp with seq_number " + sequenceNumber + " has been sent");
			} catch (PDUStringException e) {
				logger.error("Failed sending deliver_sm_resp", e);
			}
		}
		
		public void sendEnquireLinkResp(int sequenceNumber) throws IOException {
		    logger.debug("Sending enquire_link_resp");
			pduSender.sendEnquireLinkResp(out, sequenceNumber);
		}
		
		public void sendGenerickNack(int commandStatus, int sequenceNumber) throws IOException {
			pduSender.sendGenericNack(out, commandStatus, sequenceNumber);
		}
		
		public void sendNegativeResponse(int originalCommandId, int commandStatus, int sequenceNumber) throws IOException {
			pduSender.sendHeader(out, originalCommandId | SMPPConstant.MASK_CID_RESP, commandStatus, sequenceNumber);
		}
		
		public void sendUnbindResp(int sequenceNumber) throws IOException {
			pduSender.sendUnbindResp(out, SMPPConstant.STAT_ESME_ROK, sequenceNumber);
		}
	}
	
	private class PDUReaderWorker extends Thread {
		@Override
		public void run() {
			logger.info("Starting PDUReaderWorker");
			while (isReadPdu()) {
				readPDU();
			}
			close();
			logger.info("PDUReaderWorker stop");
		}
		
		private void readPDU() {
	        try {
	            Command pduHeader = null;
	            byte[] pdu = null;
	            synchronized (in) {
	                pduHeader = pduReader.readPDUHeader(in);
	                pdu = pduReader.readPDU(in, pduHeader);
	            }
	            switch (pduHeader.getCommandId()) {
	            case SMPPConstant.CID_BIND_RECEIVER_RESP:
	            case SMPPConstant.CID_BIND_TRANSMITTER_RESP:<commit_msg>Removing the synchronize block since InputStream only use by single thread.<commit_after>package org.jsmpp.session;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.SocketTimeoutException;
import java.util.Hashtable;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

import org.jsmpp.BindType;
import org.jsmpp.DefaultPDUReader;
import org.jsmpp.DefaultPDUSender;
import org.jsmpp.InterfaceVersion;
import org.jsmpp.InvalidCommandLengthException;
import org.jsmpp.InvalidResponseException;
import org.jsmpp.NumberingPlanIndicator;
import org.jsmpp.PDUReader;
import org.jsmpp.PDUSender;
import org.jsmpp.PDUStringException;
import org.jsmpp.SMPPConstant;
import org.jsmpp.SynchronizedPDUSender;
import org.jsmpp.TypeOfNumber;
import org.jsmpp.bean.BindResp;
import org.jsmpp.bean.Command;
import org.jsmpp.bean.DataCoding;
import org.jsmpp.bean.DeliverSm;
import org.jsmpp.bean.ESMClass;
import org.jsmpp.bean.EnquireLinkResp;
import org.jsmpp.bean.QuerySmResp;
import org.jsmpp.bean.RegisteredDelivery;
import org.jsmpp.bean.SubmitSmResp;
import org.jsmpp.bean.UnbindResp;
import org.jsmpp.extra.NegativeResponseException;
import org.jsmpp.extra.PendingResponse;
import org.jsmpp.extra.ProcessRequestException;
import org.jsmpp.extra.ResponseTimeoutException;
import org.jsmpp.extra.SessionState;
import org.jsmpp.session.connection.Connection;
import org.jsmpp.session.connection.ConnectionFactory;
import org.jsmpp.session.connection.socket.SocketConnectionFactory;
import org.jsmpp.session.state.SMPPSessionState;
import org.jsmpp.util.DefaultComposer;
import org.jsmpp.util.IntUtil;
import org.jsmpp.util.Sequence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author uudashr
 * @version 2.0
 *
 */
public class SMPPSession {
	private static final Logger logger = LoggerFactory.getLogger(SMPPSession.class);
	private static final Random random = new Random();
	/* Utility */
	private final PDUSender pduSender;
    private final PDUReader pduReader;
	
    /* Connection */
	private final ConnectionFactory connFactory;
	private Connection conn;
	private DataInputStream in;
	private OutputStream out;
	
	private SMPPSessionState stateProcessor = SMPPSessionState.CLOSED;
	private final Sequence sequence = new Sequence(1);
	private final ResponseHandler responseHandler = new ResponseHandlerImpl();
	private final Hashtable<Integer, PendingResponse<? extends Command>> pendingResponse = new Hashtable<Integer, PendingResponse<? extends Command>>();
	private int sessionTimer = 5000;
	private long transactionTimer = 2000;
	private long lastActivityTimestamp;
	private MessageReceiverListener messageReceiverListener;
	private SessionStateListener sessionStateListener;
    
	private EnquireLinkSender enquireLinkSender;
	private String sessionId = generateSessionId();
	
	public SMPPSession() {
        this(new SynchronizedPDUSender(new DefaultPDUSender(new DefaultComposer())), 
                new DefaultPDUReader(), 
                SocketConnectionFactory.getInstance());
    }
	
	public SMPPSession(PDUSender pduSender, PDUReader pduReader, ConnectionFactory connFactory) {
	    this.pduSender = pduSender;
	    this.pduReader = pduReader;
	    this.connFactory = connFactory;
	    
    }
	
	public SMPPSession(String host, int port, BindParameter bindParam,
            PDUSender pduSender, PDUReader pduReader,
            ConnectionFactory connFactory) throws IOException {
	    this(pduSender, pduReader, connFactory);
	    connectAndBind(host, port, bindParam);
	}
	
	public SMPPSession(String host, int port, BindParameter bindParam) throws IOException {
        this();
        connectAndBind(host, port, bindParam);
    }
	
	/**
	 * Open connection and bind immediately.
	 * 
	 * @param host is the SMSC host address.
	 * @param port is the SMSC listen port.
	 * @param bindType is the bind type.
	 * @param systemId is the system id.
	 * @param password is the password.
	 * @param systemType is the system type.
	 * @param addrTon is the address TON.
	 * @param addrNpi is the address NPI.
	 * @param addressRange is the address range.
	 * @throws IOException if there is an IO error found.
	 */
	public void connectAndBind(String host, int port, BindType bindType,
            String systemId, String password, String systemType,
            TypeOfNumber addrTon, NumberingPlanIndicator addrNpi,
            String addressRange) throws IOException {
	    connectAndBind(host, port, new BindParameter(bindType, systemId,
                password, systemType, addrTon, addrNpi, addressRange));
	}
	
	/**
	 * Open connection and bind immediately.
	 * 
	 * @param host is the SMSC host address.
	 * @param port is the SMSC listen port.
	 * @param bindParam is the bind parameters.
	 * @return the SMSC system id.
	 * @throws IOException if there is an IO error found.
	 */
	public String connectAndBind(String host, int port, BindParameter bindParam) throws IOException {
	    logger.debug("Connect and bind to " + host + " port " + port);
		if (sequence.currentValue() != 1) {
			throw new IOException("Failed connecting");
		}
		
		conn = connFactory.createConnection(host, port);
		logger.info("Connected");
		
		changeState(SessionState.OPEN);
		try {
			in = new DataInputStream(conn.getInputStream());
			out = conn.getOutputStream();
			
			new PDUReaderWorker().start();
			String smscSystemId = sendBind(bindParam.getBindType(), bindParam.getSystemId(), bindParam.getPassword(), bindParam.getSystemType(),
                    InterfaceVersion.IF_34, bindParam.getAddrTon(), bindParam.getAddrNpi(), bindParam.getAddressRange());
			changeToBoundState(bindParam.getBindType());
			
			enquireLinkSender = new EnquireLinkSender();
			enquireLinkSender.start();
			return smscSystemId;
		} catch (PDUStringException e) {
		    logger.error("Failed sending bind command", e);
		    throw new IOException("Failed sending bind since some string parameter area invalid : " + e.getMessage());
		} catch (NegativeResponseException e) {
			String message = "Receive negative bind response";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (InvalidResponseException e) {
			String message = "Receive invalid response of bind";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (ResponseTimeoutException e) {
			String message = "Waiting bind response take time to long";
			logger.error(message, e);
			close();
			throw new IOException(message + ": " + e.getMessage());
		} catch (IOException e) {
			logger.error("IO Error occur", e);
			close();
			throw e;
		}
	}
	
	/**
	 * Sending bind with 1 minutes timeout.
	 *  
	 * @param bindType is the bind type.
     * @param systemId is the system id.
     * @param password is the password.
     * @param systemTypeis the system type.
     * @param interfaceVersion is the interface version.
     * @param addrTon is the address TON.
     * @param addrNpi is the address NPI.
     * @param addressRange is the address range.
     * @return SMSC system id.
     * @throws PDUStringException if we enter invalid bind parameter(s).
     * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
     * @throws InvalidResponseException if there is invalid response found.
     * @throws NegativeResponseException if we receive negative response.
     * @throws IOException if there is an IO error occur.
	 */
	private String sendBind(BindType bindType, String systemId,
            String password, String systemType,
            InterfaceVersion interfaceVersion, TypeOfNumber addrTon,
            NumberingPlanIndicator addrNpi, String addressRange)
            throws PDUStringException, ResponseTimeoutException,
            InvalidResponseException, NegativeResponseException, IOException {
	    return sendBind(bindType, systemId, password, systemType, interfaceVersion, addrTon, addrNpi, addressRange, 60000);
	}
	
	/**
	 * Sending bind.
	 * 
	 * @param bindType is the bind type.
	 * @param systemId is the system id.
	 * @param password is the password.
	 * @param systemTypeis the system type.
	 * @param interfaceVersion is the interface version.
	 * @param addrTon is the address TON.
	 * @param addrNpi is the address NPI.
	 * @param addressRange is the address range.
	 * @param timeout is the max time waiting for bind response. 
	 * @return SMSC system id.
	 * @throws PDUStringException if we enter invalid bind parameter(s).
	 * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
	 * @throws InvalidResponseException if there is invalid response found.
	 * @throws NegativeResponseException if we receive negative response.
	 * @throws IOException if there is an IO error occur.
	 */
	private String sendBind(BindType bindType, String systemId,
			String password, String systemType,
			InterfaceVersion interfaceVersion, TypeOfNumber addrTon,
			NumberingPlanIndicator addrNpi, String addressRange, long timeout)
			throws PDUStringException, ResponseTimeoutException,
			InvalidResponseException, NegativeResponseException, IOException {
		int seqNum = sequence.nextValue();
		PendingResponse<BindResp> pendingResp = new PendingResponse<BindResp>(timeout);
		pendingResponse.put(seqNum, pendingResp);
		
		try {
			pduSender.sendBind(out, bindType, seqNum, systemId, password, systemType, interfaceVersion, addrTon, addrNpi, addressRange);
		} catch (IOException e) {
			logger.error("Failed sending bind command", e);
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		try {
			pendingResp.waitDone();
			logger.info("Bind response received");
		} catch (ResponseTimeoutException e) {
			pendingResponse.remove(seqNum);
			throw e;
		} catch (InvalidResponseException e) {
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK) {
			throw new NegativeResponseException(pendingResp.getResponse().getCommandStatus());
		}
		
		return pendingResp.getResponse().getSystemId();
	}

	/**
	 * Submit short message with specified parameter.
	 * 
     * @param serviceType is the service_type parameter.
     * @param sourceAddrTon is the source_addr_ton parameter.
     * @param sourceAddrNpi is the source_addr_npi parameter.
     * @param sourceAddr is the source_addr parameter.
     * @param destAddrTon is the dest_addr_ton parameter.
     * @param destAddrNpi is the dest_addr_npi parameter.
     * @param destinationAddr is the destination_addr parameter.
     * @param esmClass is the esm_class parameter.
     * @param protocoId is the protocol_id parameter.
     * @param priorityFlag is the priority_flag parameter.
     * @param scheduleDeliveryTime is the schedule_delivery_time parameter. (It is time in string format).
     * @param validityPeriod is the validity_period parameter (It is time in string format).
     * @param registeredDelivery is the registered_delivery parameter.
     * @param replaceIfPresentFlag is the replace_if_present_flag parameter
     * @param dataCoding is the data_coding parameter.
     * @param smDefaultMsgId is the sm_default_message_id parameter.
     * @param shortMessage is the short message.
     * @return the message id from SMSC.
     * @throws PDUStringException if we enter invalid bind parameter(s).
     * @throws ResponseTimeoutException if there is no valid response after defined millisecond.
     * @throws InvalidResponseException if there is invalid response found.
     * @throws NegativeResponseException if we receive negative response.
     * @throws IOException if there is an IO error found.
     */
    public String submitShortMessage(String serviceType,
    		TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
    		String sourceAddr, TypeOfNumber destAddrTon,
    		NumberingPlanIndicator destAddrNpi, String destinationAddr,
    		ESMClass esmClass, byte protocoId, byte priorityFlag,
    		String scheduleDeliveryTime, String validityPeriod,
    		RegisteredDelivery registeredDelivery, byte replaceIfPresentFlag,
    		DataCoding dataCoding, byte smDefaultMsgId, byte[] shortMessage)
    		throws PDUStringException, ResponseTimeoutException,
    		InvalidResponseException, NegativeResponseException, IOException {
    	
    	int seqNum = sequence.nextValue();
    	PendingResponse<SubmitSmResp> pendingResp = new PendingResponse<SubmitSmResp>(transactionTimer);
    	pendingResponse.put(seqNum, pendingResp);
    	try {
    		pduSender.sendSubmitSm(out, seqNum, serviceType, sourceAddrTon,
    				sourceAddrNpi, sourceAddr, destAddrTon, destAddrNpi,
    				destinationAddr, esmClass, protocoId, priorityFlag,
    				scheduleDeliveryTime, validityPeriod, registeredDelivery,
    				replaceIfPresentFlag, dataCoding, smDefaultMsgId, shortMessage);
    		
    	} catch (IOException e) {
    		logger.error("Failed submit short message", e);
    		pendingResponse.remove(seqNum);
    		close();
    		throw e;
    	}
    	
    	try {
    		pendingResp.waitDone();
    		logger.debug("Submit sm response received");
    	} catch (ResponseTimeoutException e) {
    		pendingResponse.remove(seqNum);
    		logger.debug("Response timeout for submit_sm with sessionIdSequence number " + seqNum);
    		throw e;
    	} catch (InvalidResponseException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	}
    	
    	if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
    		throw new NegativeResponseException(pendingResp.getResponse().getCommandStatus());
    	
    	return pendingResp.getResponse().getMessageId();
    }

    public QuerySmResult queryShortMessage(String messageId, TypeOfNumber sourceAddrTon,
    		NumberingPlanIndicator sourceAddrNpi, String sourceAddr)
    		throws PDUStringException, ResponseTimeoutException,
    		InvalidResponseException, NegativeResponseException, IOException {
    	
    	int seqNum = sequence.nextValue();
    	PendingResponse<QuerySmResp> pendingResp = new PendingResponse<QuerySmResp>(transactionTimer);
    	pendingResponse.put(seqNum, pendingResp);
    	try {
    		pduSender.sendQuerySm(out, seqNum, messageId, sourceAddrTon, sourceAddrNpi, sourceAddr);
    		
    	} catch (IOException e) {
    		logger.error("Failed submit short message", e);
    		pendingResponse.remove(seqNum);
    		close();
    		throw e;
    	}
    	
    	try {
    		pendingResp.waitDone();
    		logger.info("Query sm response received");
    	} catch (ResponseTimeoutException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	} catch (InvalidResponseException e) {
    		pendingResponse.remove(seqNum);
    		throw e;
    	}
    	
    	QuerySmResp resp = pendingResp.getResponse();
    	if (resp.getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
    		throw new NegativeResponseException(resp.getCommandStatus());
    	
    	if (resp.getMessageId().equals(messageId)) {
    		return new QuerySmResult(resp.getFinalDate(), resp.getMessageState(), resp.getErrorCode());
    	} else {
    		// message id requested not same as the returned
    		throw new InvalidResponseException("Requested message_id doesn't match with the result");
    	}
    }

    public String getSessionId() {
    	return sessionId;
    }

    private void unbind() throws ResponseTimeoutException, InvalidResponseException, IOException {
		int seqNum = sequence.nextValue();
		PendingResponse<UnbindResp> pendingResp = new PendingResponse<UnbindResp>(transactionTimer);
		pendingResponse.put(seqNum, pendingResp);
		
		try {
			pduSender.sendUnbind(out, seqNum);
		} catch (IOException e) {
			logger.error("Failed sending unbind", e);
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		try {
			pendingResp.waitDone();
			logger.info("Unbind response received");
			changeState(SessionState.UNBOUND);
		} catch (ResponseTimeoutException e) {
			pendingResponse.remove(seqNum);
			throw e;
		} catch (InvalidResponseException e) {
			pendingResponse.remove(seqNum);
			throw e;
		}
		
		if (pendingResp.getResponse().getCommandStatus() != SMPPConstant.STAT_ESME_ROK)
			logger.warn("Receive NON-OK response of unbind");
	}

	private void changeToBoundState(BindType bindType) {
		if (bindType.equals(BindType.BIND_TX)) {
			changeState(SessionState.BOUND_TX);
		} else if (bindType.equals(BindType.BIND_RX)) {
			changeState(SessionState.BOUND_RX);
		} else if (bindType.equals(BindType.BIND_TRX)) {
			changeState(SessionState.BOUND_TRX);
		} else {
			throw new IllegalArgumentException("Bind type " + bindType + " not supported");
		}
		
		try {
			conn.setSoTimeout(sessionTimer);
		} catch (IOException e) {
			logger.error("Failed setting so_timeout for session timer", e);
		}
	}

	private synchronized void changeState(SessionState newState) {
        if (!stateProcessor.getSessionState().equals(newState)) {
            final SessionState oldState = stateProcessor.getSessionState();
    		
    		// change the session state processor
    		if (newState == SessionState.OPEN) {
    			stateProcessor = SMPPSessionState.OPEN;
    		} else if (newState == SessionState.BOUND_RX) {
    			stateProcessor = SMPPSessionState.BOUND_RX;
    		} else if (newState == SessionState.BOUND_TX) {
    			stateProcessor = SMPPSessionState.BOUND_TX;
    		} else if (newState == SessionState.BOUND_TRX) {
    			stateProcessor = SMPPSessionState.BOUND_TRX;
    		} else if (newState == SessionState.UNBOUND) {
    			stateProcessor = SMPPSessionState.UNBOUND;
    		} else if (newState == SessionState.CLOSED) {
    			stateProcessor = SMPPSessionState.CLOSED;
    		}
            fireChangeState(newState, oldState);
        }
	}
	
	public int getSessionTimer() {
		return sessionTimer;
	}
	
	public void setSessionTimer(int sessionTimer) {
		this.sessionTimer = sessionTimer;
		if (stateProcessor.getSessionState().isBound()) {
			try {
				conn.setSoTimeout(sessionTimer);
			} catch (IOException e) {
				logger.error("Failed setting so_timeout for session timer", e);
			}
		}
	}
	
	public long getTransactionTimer() {
		return transactionTimer;
	}
	
	public void setTransactionTimer(long transactionTimer) {
		this.transactionTimer = transactionTimer;
	}
	
	public synchronized SessionState getSessionState() {
		return stateProcessor.getSessionState();
	}
	
	public SessionStateListener getSessionStateListener() {
        return sessionStateListener;
    }
	
    public void setSessionStateListener(
            SessionStateListener sessionStateListener) {
        this.sessionStateListener = sessionStateListener;
    }
    
    public MessageReceiverListener getMessageReceiverListener() {
        return messageReceiverListener;
    }
    
	public void setMessageReceiverListener(
			MessageReceiverListener messageReceiverListener) {
		this.messageReceiverListener = messageReceiverListener;
	}
	
    /**
     * This method provided for monitoring need.
     * 
     * @return the last activity timestamp.
     */
    public long getLastActivityTimestamp() {
        return lastActivityTimestamp;
    }

    public void unbindAndClose() {
    	try {
    		unbind();
    	} catch (ResponseTimeoutException e) {
    		logger.error("Timeout waiting unbind response", e);
    	} catch (InvalidResponseException e) {
    		logger.error("Receive invalid unbind response", e);
    	} catch (IOException e) {
    		logger.error("IO error found ", e);
    	}
    	close();
    }

    public void close() {
		changeState(SessionState.CLOSED);
		try {
            conn.close();
        } catch (IOException e) {
        }
	}
	
	private synchronized boolean isReadPdu() {
		return stateProcessor.getSessionState().isBound() || stateProcessor.getSessionState().equals(SessionState.OPEN);
	}
	
	@Override
	protected void finalize() throws Throwable {
		close();
	}
	
	private void fireAcceptDeliverSm(DeliverSm deliverSm) throws ProcessRequestException {
		if (messageReceiverListener != null) {
			messageReceiverListener.onAcceptDeliverSm(deliverSm);
        } else { 
			logger.warn("Receive deliver_sm but MessageReceiverListener is null. Short message = " + new String(deliverSm.getShortMessage()));
        }
	}
	
    private void fireChangeState(SessionState newState, SessionState oldState) {
        if (sessionStateListener != null) {
            sessionStateListener.onStateChange(newState, oldState, this);
        } else {
            logger.warn("SessionStateListener is null");
        }
    }
    
    private synchronized static final String generateSessionId() {
        return IntUtil.toHexString(random.nextInt());
    }
    
	private class ResponseHandlerImpl implements ResponseHandler {
		
		public void processDeliverSm(DeliverSm deliverSm) throws ProcessRequestException {
			fireAcceptDeliverSm(deliverSm);
		}
		
		@SuppressWarnings("unchecked")
		public PendingResponse<Command> removeSentItem(int sequenceNumber) {
			return (PendingResponse<Command>)pendingResponse.remove(sequenceNumber);
		}
		
		public void notifyUnbonded() {
		    changeState(SessionState.UNBOUND);
		}
		
		public void sendDeliverSmResp(int sequenceNumber) throws IOException {
			try {
				pduSender.sendDeliverSmResp(out, sequenceNumber);
				logger.debug("deliver_sm_resp with seq_number " + sequenceNumber + " has been sent");
			} catch (PDUStringException e) {
				logger.error("Failed sending deliver_sm_resp", e);
			}
		}
		
		public void sendEnquireLinkResp(int sequenceNumber) throws IOException {
		    logger.debug("Sending enquire_link_resp");
			pduSender.sendEnquireLinkResp(out, sequenceNumber);
		}
		
		public void sendGenerickNack(int commandStatus, int sequenceNumber) throws IOException {
			pduSender.sendGenericNack(out, commandStatus, sequenceNumber);
		}
		
		public void sendNegativeResponse(int originalCommandId, int commandStatus, int sequenceNumber) throws IOException {
			pduSender.sendHeader(out, originalCommandId | SMPPConstant.MASK_CID_RESP, commandStatus, sequenceNumber);
		}
		
		public void sendUnbindResp(int sequenceNumber) throws IOException {
			pduSender.sendUnbindResp(out, SMPPConstant.STAT_ESME_ROK, sequenceNumber);
		}
	}
	
	private class PDUReaderWorker extends Thread {
		@Override
		public void run() {
			logger.info("Starting PDUReaderWorker");
			while (isReadPdu()) {
				readPDU();
			}
			close();
			logger.info("PDUReaderWorker stop");
		}
		
		private void readPDU() {
	        try {
	            Command pduHeader = null;
	            byte[] pdu = null;
	            
                pduHeader = pduReader.readPDUHeader(in);
                pdu = pduReader.readPDU(in, pduHeader);
	            
	            switch (pduHeader.getCommandId()) {
	            case SMPPConstant.CID_BIND_RECEIVER_RESP:
	            case SMPPConstant.CID_BIND_TRANSMITTER_RESP:
{'fraction_non_alphanumeric': 0.08585393652463616, 'fraction_numerical': 0.0009644046992810802, 'mean_word_length': 3.809719586759435, 'pattern_counts': {'":': 6, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2005896', 'n_tokens_mistral': 13872, 'n_tokens_neox': 12765, 'n_words': 3361}