connection.java

SMPP（点到点短消息协议）的java实现

    private void processOutboundBind(Bind bindRequest) throws IOException {
        if (state != UNBOUND) {
            throw new IllegalStateException(
                    "Cannot bind while in state " + state);
        }
        
        // Initialise the link timeout to the bind timeout
        try {
            int bindTimeout = APIConfig.getInstance().getInt(
                    APIConfig.BIND_TIMEOUT, 0);
            if (bindTimeout > 0) {
                link.setTimeout(bindTimeout);
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Set bind timeout to " + bindTimeout);
               }
            }
        } catch (UnsupportedOperationException x) {
            LOGGER.warn("Link does not support read timeouts - bind timeout will not work");
        }
        openLink();
        setState(BINDING);
        if (asyncComms) {
            if (rcvThread == null) {
                createRecvThread();
            }
            if (!rcvThread.isAlive()) {
                rcvThread.start();
            }
        }
    }
    
    private void processOutboundUnbind(Unbind unbind) {
        if (!asyncComms && packetQueue.size() > 0) {
            throw new IllegalStateException(
                    "Cannot unbind while there are incoming packets awaiting responses");
        }
        if (state != BOUND) {
            throw new IllegalStateException("Not currently bound");
        }
        setState(UNBINDING);
    }
    
    private void processOutboundUnbindResp(UnbindResp unbindResp) {
        if (unbindResp.getCommandStatus() == 0) {
            setState(UNBOUND);
        }
    }
    
    private void processInboundPacket(SMPPPacket packet) throws IOException {
        switch (packet.getCommandId()) {
        case SMPPPacket.BIND_TRANSMITTER_RESP:
        case SMPPPacket.BIND_RECEIVER_RESP:
        case SMPPPacket.BIND_TRANSCEIVER_RESP:
            processInboundBindResp((BindResp) packet);
            break;
        case SMPPPacket.UNBIND_RESP:
            processInboundUnbindResp((UnbindResp) packet);
            break;
        case SMPPPacket.UNBIND:
            processInboundUnbind((Unbind) packet);
            break;
        case SMPPPacket.DELIVER_SM:
            if (ackDeliverSm) {
                ackDeliverSm((DeliverSM) packet);
            }
            break;
        case SMPPPacket.ENQUIRE_LINK:
            if (ackQryLinks) {
                ackEnquireLink((EnquireLink) packet);
            }
            break;
        }
        if (packet.getCommandStatus() == 0) {
            // Fix up the alphabet for this connection type if the
            // packet needs it. DCS value 0 means the alphabet is in the
            // default encoding of the SMSC, which varies depending on
            // implementation.
            if (defaultAlphabet != null && packet.getDataCoding() == 0) {
                packet.setAlphabet(defaultAlphabet);
            }
        }
    }
    
    /**
     * Handle an incoming bind response packet.
     */
    private void processInboundBindResp(BindResp resp) {
        int st = resp.getCommandStatus();

        // Throw an exception if we're not in a BINDING state..
        if (state != BINDING) {
            throw new IllegalStateException(
                    "A bind response was received in bound state " + state);
        }
        if (st != 0) {
            // Bind failed. Close the network link and return.
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Bind failed. Setting state to unbound.");
            }
            try {
                setState(UNBOUND);
                link.close();
            } catch (IOException x) {
                LOGGER.warn("I/O Exception shutting down link after failed bind.", x);
            }
            return;
        }

        // Alright so, we're bound to the SMSC..
        setState(BOUND);

        // Read the version of the protocol supported at the SMSC.
        Number n = (Number) resp.getOptionalParameter(Tag.SC_INTERFACE_VERSION);
        if (n != null) {
            SMPPVersion smscVersion = SMPPVersion.getVersion(n.intValue());
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("SMSC reports its supported SMPP version as "
                        + smscVersion.toString());
            }

            // Downgrade this connection's version if the SMSC's version is
            // lower.
            if (smscVersion.isOlder(this.interfaceVersion)) {
                LOGGER.info("Downgrading this connection's SMPP version to "
                        + smscVersion.toString());
                setInterfaceVersion(smscVersion);
            }
        } else {
            // Spec requires us to assume the SMSC does not support optional
            // parameters
            this.supportOptionalParams = false;
            LOGGER.warn("Disabling optional parameter support as no sc_interface_version parameter was received");
        }

        // Set the link timeout
        try {
            int linkTimeout = APIConfig.getInstance().getInt(
                    APIConfig.LINK_TIMEOUT);
            link.setTimeout(linkTimeout);

            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Set the link timeout to " + linkTimeout);
            }
        } catch (PropertyNotFoundException x) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("No link timeout specified in configuration");
            }
        } catch (java.lang.UnsupportedOperationException x) {
            LOGGER.warn("Configuration specified a link timeout but the link implementation does not support it");
        }
    }

    /**
     * Handle an incoming unbind packet.
     */
    private void processInboundUnbind(Unbind req) {
        LOGGER.info("SMSC requested unbind");
        setState(UNBINDING);
    }

    /**
     * Handle an incoming unbind response packet.
     */
    private void processInboundUnbindResp(UnbindResp resp) {
        try {
            if (state == UNBINDING && resp.getCommandStatus() == 0) {
                LOGGER.info("Successfully unbound");
                setState(UNBOUND);
                this.link.close();
            }
        } catch (IOException x) {
        }
    }

    /**
     * Set the event dispatcher for this connection object. Before using the new
     * event dispatcher, this method will call {@link EventDispatcher#init}to
     * initialise the dispatcher. It will then iterate through all the observers
     * registered with the current event dispatcher and register them with the
     * new one.
     * <p>
     * It is not a particularly good idea to set the event dispatcher after
     * communications have begun. However, the observer copy is guarded against
     * multi-threaded access to the current event dispatcher. During the copy,
     * however, events will continue to be delievered via the current
     * dispatcher. Only <b>after </b> the copy is complete will the new event
     * dispatcher become the active one and events begin being delivered by it.
     * </p>
     * <p>
     * The caller of this method can be sure that, once this method returns, all
     * new events will be handled by the new event dispatcher. However, there
     * may be events that occurred before, or during the operation of, the call
     * to this method which will be delivered by the old dispatcher. Once the
     * new event dispatcher is in place, the {@link EventDispatcher#destroy}
     * method will be called on the old dispatcher.
     * </p>
     */
    public void setEventDispatcher(EventDispatcher eventDispatcher) {
        if (eventDispatcher == null) {
            throw new NullPointerException("Event dispatcher cannot be null");
        }

        eventDispatcher.init();

        // Copy all current observers to the new event dispatcher..
        synchronized (this.eventDispatcher) {
            Iterator iter = this.eventDispatcher.observerIterator();
            while (iter.hasNext()) {
                eventDispatcher.addObserver((ConnectionObserver) iter.next());
            }
        }

        EventDispatcher old = this.eventDispatcher;

        // ..and swap out the old dispatcher.
        this.eventDispatcher = eventDispatcher;

        // Clean up the old dispatcher.
        old.destroy();
    }

    /**
     * Add a connection observer to receive SMPP events from this connection. If
     * this connection is not using asynchronous communication, this method call
     * has no effect.
     * 
     * @param ob
     *            the ConnectionObserver implementation to add.
     */
    public void addObserver(ConnectionObserver ob) {
        if (eventDispatcher != null) {
            eventDispatcher.addObserver(ob);
        }
    }

    /**
     * Remove a connection observer from this Connection.
     */
    public void removeObserver(ConnectionObserver ob) {
        if (eventDispatcher != null) {
            eventDispatcher.removeObserver(ob);
        }
    }

    /**
     * Listener thread method for asynchronous communication.
     */
    public void run() {
        SMPPPacket pak = null;
        int smppEx = 0;
        SMPPEvent exitEvent = null;
        int tooManyIOEx = 5;

        LOGGER.info("Receiver thread started");

        APIConfig cfg = APIConfig.getInstance();
        try {
            tooManyIOEx = cfg.getInt(APIConfig.TOO_MANY_IO_EXCEPTIONS);
        } catch (PropertyNotFoundException x) {
            // just stick with the default
            LOGGER.debug("Didn't find I/O exception config. Using default of "
                    + tooManyIOEx);
        }

        eventDispatcher.notifyObservers(this, new ReceiverStartEvent(this));
        try {
            while (state != UNBOUND) {
                try {
                    pak = readNextPacketInternal();
                    if (pak == null) {
                        LOGGER.warn("Received an unidentified packet from the SMSC");
                        continue;
                   }
               } catch (SocketTimeoutException x) {
                   if (LOGGER.isDebugEnabled()) {
                       LOGGER.debug("Caught a socket timeout exception: "
                               + x.getMessage());
                   }
                   if (state == BINDING) {
                       // bind timeout has expired
                       LOGGER.debug("Bind timeout.");
                       exitEvent = new ReceiverExitEvent(this, null, state);
                       ((ReceiverExitEvent) exitEvent).setReason(
                               ReceiverExitEvent.BIND_TIMEOUT);
                       setState(UNBOUND);
                   } else {
                       eventDispatcher.notifyObservers(this,
                               new ReceiverExceptionEvent(this, x));
                   }
                   continue;
               } catch (IOException x) {
                   LOGGER.warn("I/O Exception caught", x);
                   ReceiverExceptionEvent ev = new ReceiverExceptionEvent(
                           this, x, state);
                   eventDispatcher.notifyObservers(this, ev);
                   smppEx++;
                   if (smppEx > tooManyIOEx) {
                       LOGGER.warn("Too many IOExceptions in receiver thread", x);
                       throw x;
                   }
                   continue;
               }

                // Reset smppEx back to zero if we reach here, as packet
                // reception was successful.
                smppEx = 0;

                // Tell all the observers about the new packet
                LOGGER.info("Notifying observers of packet received");
                eventDispatcher.notifyObservers(this, pak);
            } // end while
            if (exitEvent == null) {
                // Notify observers that the thread is exiting with no error..
                exitEvent = new ReceiverExitEvent(this, null, state);
            }
        } catch (Exception x) {
            LOGGER.debug("Fatal exception in receiver thread: " + x.getMessage(), x);
            exitEvent = new ReceiverExitEvent(this, x, state);
            setState(UNBOUND);
        } finally {
            // make sure other code doesn't try to restart the rcvThread..
            rcvThread = null;
        }

        if (exitEvent != null) {
            eventDispatcher.notifyObservers(this, exitEvent);
        }

        // Clean up the event dispatcher.
        eventDispatcher.destroy();
    }

    /**
     * @deprecated #ackEnquireLink
     */
    public void ackLinkQuery(EnquireLink el) throws java.io.IOException {
        ackEnquireLink(el);
    }

    /**
     * Get a new instance of an SMPP packet. The packet will be initialised so
     * that it uses the same SMPP version as this connection and it's sequence
     * number will be initialised to using this connection's sequence numbering
     * scheme.
     * 
     * @param commandId
     *            the SMPP command ID of the packet to retrieve.
     * @return a subclass of {@link ie.omk.smpp.message.SMPPPacket}
     *         corresponding to SMPP command <code>commandId</code>.
     * @throws ie.omk.smpp.BadCommandIDException
     *             if the command ID is not recognised.
     * @throws ie.omk.smpp.NotSupportedException
     *             if the Connection is currently using an SMPP version which
     *             does not support SMPP command <code>commandId</code>.
     */
    public SMPPPacket newInstance(int commandId) throws BadCommandIDException,
            VersionException {

        if (!this.interfaceVersion.isSupported(commandId)) {
            throw new VersionException(
                    "Command is not supported in this SMPP version");
        }

        SMPPPacket response = PacketFactory.newInstance(commandId);
        response.setVersion(this.interfaceVersion);
        if (this.seqNumScheme != null) {
            response.setSequenceNum(this.seqNumScheme.nextNumber());
        }

        if (defaultAlphabet != null) {
            response.setAlphabet(defaultAlphabet, 0);
        }

        return response;
    }
}