<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.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.jsmpp.DefaultPDUReader;
import org.jsmpp.DefaultPDUSender;
import org.jsmpp.InvalidCommandLengthException;
import org.jsmpp.InvalidResponseException;
import org.jsmpp.PDUReader;
import org.jsmpp.PDUSender;
import org.jsmpp.PDUStringException;
import org.jsmpp.SMPPConstant;
import org.jsmpp.SynchronizedPDUSender;
import org.jsmpp.bean.Bind;
import org.jsmpp.bean.BindType;
import org.jsmpp.bean.CancelSm;
import org.jsmpp.bean.Command;
import org.jsmpp.bean.DataCoding;
import org.jsmpp.bean.DataSm;
import org.jsmpp.bean.ESMClass;
import org.jsmpp.bean.MessageState;
import org.jsmpp.bean.NumberingPlanIndicator;
import org.jsmpp.bean.OptionalParameter;
import org.jsmpp.bean.QuerySm;
import org.jsmpp.bean.RegisteredDelivery;
import org.jsmpp.bean.ReplaceSm;
import org.jsmpp.bean.SubmitMulti;
import org.jsmpp.bean.SubmitMultiResult;
import org.jsmpp.bean.SubmitSm;
import org.jsmpp.bean.TypeOfNumber;
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.util.MessageId;
import org.jsmpp.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author uudashr
 *
 */
public class SMPPServerSession extends AbstractSession implements ServerSession {
    private static final Logger logger = LoggerFactory.getLogger(SMPPServerSession.class);
    
    private final Connection conn;
    private final DataInputStream in;
    private final OutputStream out;
    
    private final PDUReader pduReader;
    
    private SMPPServerSessionContext sessionContext = new SMPPServerSessionContext(this);
    private final ServerResponseHandler responseHandler = new ResponseHandlerImpl();
    
    private ServerMessageReceiverListener messageReceiverListener;
    private final EnquireLinkSender enquireLinkSender;
    private BindRequestReceiver bindRequestReceiver = new BindRequestReceiver(responseHandler);
    
    public SMPPServerSession(Connection conn,
            SessionStateListener sessionStateListener,
            ServerMessageReceiverListener messageReceiverListener,
            int pduProcessorDegree) {
        this(conn, sessionStateListener, messageReceiverListener,
                pduProcessorDegree, new SynchronizedPDUSender(
                        new DefaultPDUSender()), new DefaultPDUReader());
    }
    
    public SMPPServerSession(Connection conn,
            SessionStateListener sessionStateListener,
            ServerMessageReceiverListener messageReceiverListener,
            int pduProcessorDegree, PDUSender pduSender, PDUReader pduReader) {
        super(pduSender);
        this.conn = conn;
        this.messageReceiverListener = messageReceiverListener;
        this.pduReader = pduReader;
        this.in = new DataInputStream(conn.getInputStream());
        this.out = conn.getOutputStream();
        enquireLinkSender = new EnquireLinkSender();
        addSessionStateListener(new BoundStateListener());
        addSessionStateListener(sessionStateListener);
        setPduProcessorDegree(pduProcessorDegree);
        sessionContext.open();
    }
    
    /**
     * Wait for bind request.
     * 
     * @param timeout is the timeout.
     * @return the {@link BindRequest}.
     * @throws IllegalStateException if this invocation of this method has been
     *         made or invoke when state is not OPEN.
     * @throws TimeoutException if the timeout has been reach and
     *         {@link SMPPServerSession} are no more valid because the
     *         connection will be close automatically.
     */
    public BindRequest waitForBind(long timeout) throws IllegalStateException, TimeoutException {
        if (getSessionState().equals(SessionState.OPEN)) {
            new PDUReaderWorker().start();
            try {
                return bindRequestReceiver.waitForRequest(timeout);
            } catch (IllegalStateException e) {
                throw new IllegalStateException("Invocation of waitForBind() has been made", e);
            } catch (TimeoutException e) {
                close();
                throw e;
            }
        } else {
            throw new IllegalStateException("waitForBind() should be invoked on OPEN state, actual state is " + SessionState.OPEN);
        }
    }
    
    private synchronized boolean isReadPdu() {
        SessionState sessionState = sessionContext.getSessionState();
        return sessionState.isBound() || sessionState.equals(SessionState.OPEN);
    }
    
    public void deliverShortMessage(String serviceType,
            TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
            String sourceAddr, TypeOfNumber destAddrTon,
            NumberingPlanIndicator destAddrNpi, String destinationAddr,
            ESMClass esmClass, byte protocoId, byte priorityFlag,
            RegisteredDelivery registeredDelivery, DataCoding dataCoding,
            byte[] shortMessage, OptionalParameter... optionalParameters)
            throws PDUStringException, ResponseTimeoutException,
            InvalidResponseException, NegativeResponseException, IOException {
        
        DeliverSmCommandTask task = new DeliverSmCommandTask(pduSender(),
                serviceType, sourceAddrTon, sourceAddrNpi, sourceAddr,
                destAddrTon, destAddrNpi, destinationAddr, esmClass, protocoId,
                protocoId, registeredDelivery, dataCoding, shortMessage,
                optionalParameters);
        
        executeSendCommand(task, getTransactionTimer());
    }
    
    /* (non-Javadoc)
     * @see org.jsmpp.session.ServerSession#alertNotification(int, org.jsmpp.bean.TypeOfNumber, org.jsmpp.bean.NumberingPlanIndicator, java.lang.String, org.jsmpp.bean.TypeOfNumber, org.jsmpp.bean.NumberingPlanIndicator, java.lang.String, org.jsmpp.bean.OptionalParameter[])
     */
    public void alertNotification(int sequenceNumber,
            TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
            String sourceAddr, TypeOfNumber esmeAddrTon,
            NumberingPlanIndicator esmeAddrNpi, String esmeAddr,
            OptionalParameter... optionalParameters) throws PDUStringException,
            IOException {
        pduSender().sendAlertNotification(connection().getOutputStream(),
                sequenceNumber, sourceAddrTon.value(), sourceAddrNpi.value(),
                sourceAddr, esmeAddrTon.value(), esmeAddrNpi.value(), esmeAddr,
                optionalParameters);
    }
    
    private MessageId fireAcceptSubmitSm(SubmitSm submitSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptSubmitSm(submitSm, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private SubmitMultiResult fireAcceptSubmitMulti(SubmitMulti submitMulti) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptSubmitMulti(submitMulti, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private QuerySmResult fireAcceptQuerySm(QuerySm querySm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptQuerySm(querySm, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private void fireAcceptReplaceSm(ReplaceSm replaceSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            messageReceiverListener.onAcceptReplaceSm(replaceSm, this);
        } else {
            throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
        }
    }
    
    private void fireAcceptCancelSm(CancelSm cancelSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            messageReceiverListener.onAcceptCancelSm(cancelSm, this);
        } else {
            throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
        }
    }
    
    @Override
    protected Connection connection() {
        return conn;
    }
    
    @Override
    protected AbstractSessionContext sessionContext() {
        return sessionContext;
    }
    
    @Override
    protected GenericMessageReceiverListener messageReceiverListener() {
        return messageReceiverListener;
    }
    
    public ServerMessageReceiverListener getMessageReceiverListener() {
        return messageReceiverListener;
    }
    
    public void setMessageReceiverListener(
            ServerMessageReceiverListener messageReceiverListener) {
        this.messageReceiverListener = messageReceiverListener;
    }
    
    private class ResponseHandlerImpl implements ServerResponseHandler {
        @SuppressWarnings("unchecked")
        public PendingResponse<Command> removeSentItem(int sequenceNumber) {
            return removePendingResponse(sequenceNumber);
        }
        
        public void notifyUnbonded() {
            sessionContext.unbound();
        }
        
        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);
        }
        
        public void sendBindResp(String systemId, BindType bindType, int sequenceNumber) throws IOException {
            sessionContext.bound(bindType);
            try {
                pduSender().sendBindResp(out, bindType.responseCommandId(), sequenceNumber, systemId);
            } catch (PDUStringException e) {
                logger.error("Failed sending bind response", e);
                // TODO uudashr: we have double checking when accept the bind request
            }
        }
        
        public void processBind(Bind bind) {
            bindRequestReceiver.notifyAcceptBind(bind);
        }
        
        public MessageId processSubmitSm(SubmitSm submitSm)
                throws ProcessRequestException {
            return fireAcceptSubmitSm(submitSm);
        }
        
        public void sendSubmitSmResponse(MessageId messageId, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendSubmitSmResp(out, sequenceNumber,
                        messageId.getValue());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of MessageId should be save.
                 */
                logger.error("SYSTEM ERROR. Failed sending submitSmResp", e);
            }
        }
        
        public SubmitMultiResult processSubmitMulti(SubmitMulti submitMulti)
                throws ProcessRequestException {
            return fireAcceptSubmitMulti(submitMulti);
        }
        
        public void sendSubmitMultiResponse(
                SubmitMultiResult submiitMultiResult, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendSubmitMultiResp(out, sequenceNumber,
                        submiitMultiResult.getMessageId(),
                        submiitMultiResult.getUnsuccessDeliveries());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of the response parameter has been validated.
                 */
                logger.error("SYSTEM ERROR. Failed sending submitMultiResp", e);
            }
        }
        
        public QuerySmResult processQuerySm(QuerySm querySm)
                throws ProcessRequestException {
            return fireAcceptQuerySm(querySm);
        }
        
        public void sendQuerySmResp(String messageId, String finalDate,
                MessageState messageState, byte errorCode, int sequenceNumber) throws IOException {
            try {
                pduSender().sendQuerySmResp(out, sequenceNumber, messageId,
                        finalDate, messageState, errorCode);
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of parsed messageId has been validated.
                 */
                logger.error("SYSTEM ERROR. Failed sending cancelSmResp", e);
            }
        }
        
        public DataSmResult processDataSm(DataSm dataSm)
                throws ProcessRequestException {
            return fireAcceptDataSm(dataSm);
        }
        
        // TODO uudashr: we can generalize this method 
        public void sendDataSmResp(DataSmResult dataSmResult, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendDataSmResp(out, sequenceNumber,
                        dataSmResult.getMessageId(),
                        dataSmResult.getOptionalParameters());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of MessageId should be save.
                 */
                logger.error("SYSTEM ERROR. Failed sending dataSmResp", e);
            }
        }
        
        public void processCancelSm(CancelSm cancelSm)
                throws ProcessRequestException {
            fireAcceptCancelSm(cancelSm);
        }
        
        public void sendCancelSmResp(int sequenceNumber) throws IOException {
            pduSender().sendCancelSmResp(out, sequenceNumber);
        }
        
        
        public void processReplaceSm(ReplaceSm replaceSm)
                throws ProcessRequestException {
            fireAcceptReplaceSm(replaceSm);
        }
        
        public void sendReplaceSmResp(int sequenceNumber) throws IOException {
            pduSender().sendReplaceSmResp(out, sequenceNumber);
        }
    }
    
    private class PDUReaderWorker extends Thread {
        private StopWatch stopWatch = new StopWatch();
        private ExecutorService executorService = Executors.newFixedThreadPool(getPduProcessorDegree());
        private Runnable onIOExceptionTask = new Runnable() {
            public void run() {
                close();
            };
        };
        
        @Override
        public void run() {
            logger.info("Starting PDUReaderWorker with processor degree:{} ...", getPduProcessorDegree());
            while (isReadPdu()) {
                readPDU();
            }
            close();
            executorService.shutdown();
            logger.info("PDUReaderWorker stop");
        }
        
        private void readPDU() {
            try {
                Command pduHeader = null;
                byte[] pdu = null;

                pduHeader = pduReader.readPDUHeader(in);
                pdu = pduReader.readPDU(in, pduHeader);
                
                stopWatch.start();
                PDUProcessServerTask task = new PDUProcessServerTask(pduHeader,
                        pdu, sessionContext.getStateProcessor(),
                        sessionContext, responseHandler, onIOExceptionTask);
                executorService.execute(task);
                long delay = stopWatch.done();
                if (delay > 1000) {
                    logger.warn("Delay of processing PDU {} is to long: {}", pduHeader.getCommandIdAsHex(), delay);
                }
            } catch (InvalidCommandLengthException e) {
                logger.warn("Receive invalid command length", e);
                try {<commit_msg>Remove the StopWatch when reading pdu<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.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.jsmpp.DefaultPDUReader;
import org.jsmpp.DefaultPDUSender;
import org.jsmpp.InvalidCommandLengthException;
import org.jsmpp.InvalidResponseException;
import org.jsmpp.PDUReader;
import org.jsmpp.PDUSender;
import org.jsmpp.PDUStringException;
import org.jsmpp.SMPPConstant;
import org.jsmpp.SynchronizedPDUSender;
import org.jsmpp.bean.Bind;
import org.jsmpp.bean.BindType;
import org.jsmpp.bean.CancelSm;
import org.jsmpp.bean.Command;
import org.jsmpp.bean.DataCoding;
import org.jsmpp.bean.DataSm;
import org.jsmpp.bean.ESMClass;
import org.jsmpp.bean.MessageState;
import org.jsmpp.bean.NumberingPlanIndicator;
import org.jsmpp.bean.OptionalParameter;
import org.jsmpp.bean.QuerySm;
import org.jsmpp.bean.RegisteredDelivery;
import org.jsmpp.bean.ReplaceSm;
import org.jsmpp.bean.SubmitMulti;
import org.jsmpp.bean.SubmitMultiResult;
import org.jsmpp.bean.SubmitSm;
import org.jsmpp.bean.TypeOfNumber;
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.util.MessageId;
import org.jsmpp.util.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author uudashr
 *
 */
public class SMPPServerSession extends AbstractSession implements ServerSession {
    private static final Logger logger = LoggerFactory.getLogger(SMPPServerSession.class);
    
    private final Connection conn;
    private final DataInputStream in;
    private final OutputStream out;
    
    private final PDUReader pduReader;
    
    private SMPPServerSessionContext sessionContext = new SMPPServerSessionContext(this);
    private final ServerResponseHandler responseHandler = new ResponseHandlerImpl();
    
    private ServerMessageReceiverListener messageReceiverListener;
    private final EnquireLinkSender enquireLinkSender;
    private BindRequestReceiver bindRequestReceiver = new BindRequestReceiver(responseHandler);
    
    public SMPPServerSession(Connection conn,
            SessionStateListener sessionStateListener,
            ServerMessageReceiverListener messageReceiverListener,
            int pduProcessorDegree) {
        this(conn, sessionStateListener, messageReceiverListener,
                pduProcessorDegree, new SynchronizedPDUSender(
                        new DefaultPDUSender()), new DefaultPDUReader());
    }
    
    public SMPPServerSession(Connection conn,
            SessionStateListener sessionStateListener,
            ServerMessageReceiverListener messageReceiverListener,
            int pduProcessorDegree, PDUSender pduSender, PDUReader pduReader) {
        super(pduSender);
        this.conn = conn;
        this.messageReceiverListener = messageReceiverListener;
        this.pduReader = pduReader;
        this.in = new DataInputStream(conn.getInputStream());
        this.out = conn.getOutputStream();
        enquireLinkSender = new EnquireLinkSender();
        addSessionStateListener(new BoundStateListener());
        addSessionStateListener(sessionStateListener);
        setPduProcessorDegree(pduProcessorDegree);
        sessionContext.open();
    }
    
    /**
     * Wait for bind request.
     * 
     * @param timeout is the timeout.
     * @return the {@link BindRequest}.
     * @throws IllegalStateException if this invocation of this method has been
     *         made or invoke when state is not OPEN.
     * @throws TimeoutException if the timeout has been reach and
     *         {@link SMPPServerSession} are no more valid because the
     *         connection will be close automatically.
     */
    public BindRequest waitForBind(long timeout) throws IllegalStateException, TimeoutException {
        if (getSessionState().equals(SessionState.OPEN)) {
            new PDUReaderWorker().start();
            try {
                return bindRequestReceiver.waitForRequest(timeout);
            } catch (IllegalStateException e) {
                throw new IllegalStateException("Invocation of waitForBind() has been made", e);
            } catch (TimeoutException e) {
                close();
                throw e;
            }
        } else {
            throw new IllegalStateException("waitForBind() should be invoked on OPEN state, actual state is " + SessionState.OPEN);
        }
    }
    
    private synchronized boolean isReadPdu() {
        SessionState sessionState = sessionContext.getSessionState();
        return sessionState.isBound() || sessionState.equals(SessionState.OPEN);
    }
    
    public void deliverShortMessage(String serviceType,
            TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
            String sourceAddr, TypeOfNumber destAddrTon,
            NumberingPlanIndicator destAddrNpi, String destinationAddr,
            ESMClass esmClass, byte protocoId, byte priorityFlag,
            RegisteredDelivery registeredDelivery, DataCoding dataCoding,
            byte[] shortMessage, OptionalParameter... optionalParameters)
            throws PDUStringException, ResponseTimeoutException,
            InvalidResponseException, NegativeResponseException, IOException {
        
        DeliverSmCommandTask task = new DeliverSmCommandTask(pduSender(),
                serviceType, sourceAddrTon, sourceAddrNpi, sourceAddr,
                destAddrTon, destAddrNpi, destinationAddr, esmClass, protocoId,
                protocoId, registeredDelivery, dataCoding, shortMessage,
                optionalParameters);
        
        executeSendCommand(task, getTransactionTimer());
    }
    
    /* (non-Javadoc)
     * @see org.jsmpp.session.ServerSession#alertNotification(int, org.jsmpp.bean.TypeOfNumber, org.jsmpp.bean.NumberingPlanIndicator, java.lang.String, org.jsmpp.bean.TypeOfNumber, org.jsmpp.bean.NumberingPlanIndicator, java.lang.String, org.jsmpp.bean.OptionalParameter[])
     */
    public void alertNotification(int sequenceNumber,
            TypeOfNumber sourceAddrTon, NumberingPlanIndicator sourceAddrNpi,
            String sourceAddr, TypeOfNumber esmeAddrTon,
            NumberingPlanIndicator esmeAddrNpi, String esmeAddr,
            OptionalParameter... optionalParameters) throws PDUStringException,
            IOException {
        pduSender().sendAlertNotification(connection().getOutputStream(),
                sequenceNumber, sourceAddrTon.value(), sourceAddrNpi.value(),
                sourceAddr, esmeAddrTon.value(), esmeAddrNpi.value(), esmeAddr,
                optionalParameters);
    }
    
    private MessageId fireAcceptSubmitSm(SubmitSm submitSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptSubmitSm(submitSm, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private SubmitMultiResult fireAcceptSubmitMulti(SubmitMulti submitMulti) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptSubmitMulti(submitMulti, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private QuerySmResult fireAcceptQuerySm(QuerySm querySm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            return messageReceiverListener.onAcceptQuerySm(querySm, this);
        }
        throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
    }
    
    private void fireAcceptReplaceSm(ReplaceSm replaceSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            messageReceiverListener.onAcceptReplaceSm(replaceSm, this);
        } else {
            throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
        }
    }
    
    private void fireAcceptCancelSm(CancelSm cancelSm) throws ProcessRequestException {
        if (messageReceiverListener != null) {
            messageReceiverListener.onAcceptCancelSm(cancelSm, this);
        } else {
            throw new ProcessRequestException("MessageReceveiverListener hasn't been set yet", SMPPConstant.STAT_ESME_RX_R_APPN);
        }
    }
    
    @Override
    protected Connection connection() {
        return conn;
    }
    
    @Override
    protected AbstractSessionContext sessionContext() {
        return sessionContext;
    }
    
    @Override
    protected GenericMessageReceiverListener messageReceiverListener() {
        return messageReceiverListener;
    }
    
    public ServerMessageReceiverListener getMessageReceiverListener() {
        return messageReceiverListener;
    }
    
    public void setMessageReceiverListener(
            ServerMessageReceiverListener messageReceiverListener) {
        this.messageReceiverListener = messageReceiverListener;
    }
    
    private class ResponseHandlerImpl implements ServerResponseHandler {
        @SuppressWarnings("unchecked")
        public PendingResponse<Command> removeSentItem(int sequenceNumber) {
            return removePendingResponse(sequenceNumber);
        }
        
        public void notifyUnbonded() {
            sessionContext.unbound();
        }
        
        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);
        }
        
        public void sendBindResp(String systemId, BindType bindType, int sequenceNumber) throws IOException {
            sessionContext.bound(bindType);
            try {
                pduSender().sendBindResp(out, bindType.responseCommandId(), sequenceNumber, systemId);
            } catch (PDUStringException e) {
                logger.error("Failed sending bind response", e);
                // TODO uudashr: we have double checking when accept the bind request
            }
        }
        
        public void processBind(Bind bind) {
            bindRequestReceiver.notifyAcceptBind(bind);
        }
        
        public MessageId processSubmitSm(SubmitSm submitSm)
                throws ProcessRequestException {
            return fireAcceptSubmitSm(submitSm);
        }
        
        public void sendSubmitSmResponse(MessageId messageId, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendSubmitSmResp(out, sequenceNumber,
                        messageId.getValue());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of MessageId should be save.
                 */
                logger.error("SYSTEM ERROR. Failed sending submitSmResp", e);
            }
        }
        
        public SubmitMultiResult processSubmitMulti(SubmitMulti submitMulti)
                throws ProcessRequestException {
            return fireAcceptSubmitMulti(submitMulti);
        }
        
        public void sendSubmitMultiResponse(
                SubmitMultiResult submiitMultiResult, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendSubmitMultiResp(out, sequenceNumber,
                        submiitMultiResult.getMessageId(),
                        submiitMultiResult.getUnsuccessDeliveries());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of the response parameter has been validated.
                 */
                logger.error("SYSTEM ERROR. Failed sending submitMultiResp", e);
            }
        }
        
        public QuerySmResult processQuerySm(QuerySm querySm)
                throws ProcessRequestException {
            return fireAcceptQuerySm(querySm);
        }
        
        public void sendQuerySmResp(String messageId, String finalDate,
                MessageState messageState, byte errorCode, int sequenceNumber) throws IOException {
            try {
                pduSender().sendQuerySmResp(out, sequenceNumber, messageId,
                        finalDate, messageState, errorCode);
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of parsed messageId has been validated.
                 */
                logger.error("SYSTEM ERROR. Failed sending cancelSmResp", e);
            }
        }
        
        public DataSmResult processDataSm(DataSm dataSm)
                throws ProcessRequestException {
            return fireAcceptDataSm(dataSm);
        }
        
        // TODO uudashr: we can generalize this method 
        public void sendDataSmResp(DataSmResult dataSmResult, int sequenceNumber)
                throws IOException {
            try {
                pduSender().sendDataSmResp(out, sequenceNumber,
                        dataSmResult.getMessageId(),
                        dataSmResult.getOptionalParameters());
            } catch (PDUStringException e) {
                /*
                 * There should be no PDUStringException thrown since creation
                 * of MessageId should be save.
                 */
                logger.error("SYSTEM ERROR. Failed sending dataSmResp", e);
            }
        }
        
        public void processCancelSm(CancelSm cancelSm)
                throws ProcessRequestException {
            fireAcceptCancelSm(cancelSm);
        }
        
        public void sendCancelSmResp(int sequenceNumber) throws IOException {
            pduSender().sendCancelSmResp(out, sequenceNumber);
        }
        
        
        public void processReplaceSm(ReplaceSm replaceSm)
                throws ProcessRequestException {
            fireAcceptReplaceSm(replaceSm);
        }
        
        public void sendReplaceSmResp(int sequenceNumber) throws IOException {
            pduSender().sendReplaceSmResp(out, sequenceNumber);
        }
    }
    
    private class PDUReaderWorker extends Thread {
        private ExecutorService executorService = Executors.newFixedThreadPool(getPduProcessorDegree());
        private Runnable onIOExceptionTask = new Runnable() {
            public void run() {
                close();
            };
        };
        
        @Override
        public void run() {
            logger.info("Starting PDUReaderWorker with processor degree:{} ...", getPduProcessorDegree());
            while (isReadPdu()) {
                readPDU();
            }
            close();
            executorService.shutdown();
            logger.info("PDUReaderWorker stop");
        }
        
        private void readPDU() {
            try {
                Command pduHeader = null;
                byte[] pdu = null;

                pduHeader = pduReader.readPDUHeader(in);
                pdu = pduReader.readPDU(in, pduHeader);
                
                PDUProcessServerTask task = new PDUProcessServerTask(pduHeader,
                        pdu, sessionContext.getStateProcessor(),
                        sessionContext, responseHandler, onIOExceptionTask);
                executorService.execute(task);
            } catch (InvalidCommandLengthException e) {
                logger.warn("Receive invalid command length", e);
                try {
{'fraction_non_alphanumeric': 0.06866546304791783, 'fraction_numerical': 0.00023677745878592359, 'mean_word_length': 2.458341862845445, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, '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': '7245537', 'n_tokens_mistral': 8401, 'n_tokens_neox': 8166, 'n_words': 1868}