import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createConnectionManagerListener() {
    // Start multiplexers socket unless it's been disabled.
    if (isConnectionManagerListenerEnabled()) {
        // Create SocketAcceptor with correct number of processors
        multiplexerSocketAcceptor = buildSocketAcceptor();
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("connectionManager");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor) threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);
        multiplexerSocketAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        multiplexerSocketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));

    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createComponentListener() {
    // Start components socket unless it's been disabled.
    if (isComponentListenerEnabled() && componentAcceptor == null) {
        // Create SocketAcceptor with correct number of processors
        componentAcceptor = buildSocketAcceptor();
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("component");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.component.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);

        componentAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        componentAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createClientListeners() {
    // Start clients plain socket unless it's been disabled.
    if (isClientListenerEnabled()) {
        // Create SocketAcceptor with correct number of processors
        socketAcceptor = buildSocketAcceptor();
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("client");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.client.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);

        socketAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        socketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
        // Kill sessions whose outgoing queues keep growing and fail to send traffic
        socketAcceptor.getFilterChain().addAfter("xmpp", "outCap", new StalledSessionsFilter());
    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createConnectionManagerListener() {
    // Start multiplexers socket unless it's been disabled.
    if (isConnectionManagerListenerEnabled()) {
        // Create SocketAcceptor with correct number of processors
        multiplexerSocketAcceptor = buildSocketAcceptor("multiplexer");
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("connectionManager");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.multiplex.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor) threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);
        multiplexerSocketAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        multiplexerSocketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));

    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createComponentListener() {
    // Start components socket unless it's been disabled.
    if (isComponentListenerEnabled() && componentAcceptor == null) {
        // Create SocketAcceptor with correct number of processors
        componentAcceptor = buildSocketAcceptor("component");
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("component");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.component.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);

        componentAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        componentAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void createClientListeners() {
    // Start clients plain socket unless it's been disabled.
    if (isClientListenerEnabled()) {
        // Create SocketAcceptor with correct number of processors
        socketAcceptor = buildSocketAcceptor("client");
        // Customize Executor that will be used by processors to process incoming stanzas
        ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("client");
        int eventThreads = JiveGlobals.getIntProperty("xmpp.client.processing.threads", 16);
        ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
        eventExecutor.setCorePoolSize(eventThreads + 1);
        eventExecutor.setMaximumPoolSize(eventThreads + 1);
        eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);

        socketAcceptor.getDefaultConfig().setThreadModel(threadModel);
        // Add the XMPP codec filter
        socketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
        // Kill sessions whose outgoing queues keep growing and fail to send traffic
        socketAcceptor.getFilterChain().addAfter("xmpp", "outCap", new StalledSessionsFilter());
    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
public void accept(final NetworkTransportConfiguration config, final ProtocolEngineFactory factory,
        final SSLContextFactory sslFactory)
{
    int processors = config.getConnectorProcessors();
    
    if (Transport.TCP.equalsIgnoreCase(config.getTransport()))
    {
        _acceptor = new SocketAcceptor(processors, new NewThreadExecutor());

        SocketAcceptorConfig sconfig = (SocketAcceptorConfig) _acceptor.getDefaultConfig();
        sconfig.setThreadModel(ExecutorThreadModel.getInstance("MinaNetworkTransport(Acceptor)"));
        SocketSessionConfig sc = (SocketSessionConfig) sconfig.getSessionConfig();
        sc.setTcpNoDelay(config.getTcpNoDelay());
        sc.setSendBufferSize(config.getSendBufferSize());
        sc.setReceiveBufferSize(config.getReceiveBufferSize());

        if (config.getHost().equals(WILDCARD_ADDRESS))
        {
            _address = new InetSocketAddress(config.getPort());
        }
        else
        {
            _address = new InetSocketAddress(config.getHost(), config.getPort());
        }
    }
    else
    {
        throw new TransportException("Unknown transport: " + config.getTransport());
    }

    try
    {
        _acceptor.bind(_address, new MinaNetworkHandler(sslFactory, factory));
    }
    catch (IOException e)
    {
        throw new TransportException("Could not bind to " + _address, e);
    }
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
public NetworkConnection connect(Receiver<java.nio.ByteBuffer> receiver, ConnectionSettings settings, SSLContextFactory sslFactory)
{
    final IoConnector ioConnector = _ioConnectorFactory.newConnector();
    final SocketAddress address;
    final String protocol = settings.getProtocol();
    final int port = settings.getPort();

    if (Transport.TCP.equalsIgnoreCase(protocol))
    {
        address = new InetSocketAddress(settings.getHost(), port);
    }
    else
    {
        throw new TransportException("Unknown transport: " + protocol);
    }

    LOGGER.debug("Attempting connection to " + address);

    if (ioConnector instanceof SocketConnector)
    {
        SocketConnectorConfig cfg = (SocketConnectorConfig) ioConnector.getDefaultConfig();
        cfg.setThreadModel(ExecutorThreadModel.getInstance("MinaNetworkTransport(Client)"));

        SocketSessionConfig scfg = (SocketSessionConfig) cfg.getSessionConfig();
        scfg.setTcpNoDelay(true);
        scfg.setSendBufferSize(CLIENT_DEFAULT_BUFFER_SIZE);
        scfg.setReceiveBufferSize(CLIENT_DEFAULT_BUFFER_SIZE);

        // Don't have the connector's worker thread wait around for other
        // connections (we only use one SocketConnector per connection
        // at the moment anyway). This allows short-running
        // clients (like unit tests) to complete quickly.
        ((SocketConnector) ioConnector).setWorkerTimeout(0);
    }

    if (settings.isUseSSL()){
        try {
            SSLContext sslContext = SSLUtil.createSSLContext(settings);
            SSLFilter sslFilter = new SSLFilter(sslContext);
            sslFilter.setUseClientMode(true);
            ioConnector.getFilterChain().addFirst("sslFilter", sslFilter);
        } catch (Exception e) {
            Exception ex = new Exception("An exception occurred in creating SSLContext: ", e);
            ex.printStackTrace();
        }
    }

    ConnectFuture future = ioConnector.connect(address, new MinaNetworkHandler(null), ioConnector.getDefaultConfig());
    future.join();
    if (!future.isConnected())
    {
        throw new TransportException("Could not open connection");
    }
    IoSession session = future.getSession();
    session.setAttachment(receiver);

    return new MinaNetworkConnection(session);
}
 

import org.apache.mina.common.ExecutorThreadModel; //导入依赖的package包/类
private void startClientListeners(String localIPAddress) {
    if (!JiveGlobals.getBooleanProperty("xmpp.socket.default.active", true)) {
        // Do not start listener if service is disabled
        return;
    }
    // Start clients plain socket unless it's been disabled.
    int port = getClientListenerPort();
    // Create SocketAcceptor with correct number of processors
    socketAcceptor = buildSocketAcceptor();
    // Customize Executor that will be used by processors to process incoming stanzas
    ExecutorThreadModel threadModel = ExecutorThreadModel.getInstance("client");
    int eventThreads = JiveGlobals.getIntProperty("xmpp.processor.threads.standard", 16);
    ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor)threadModel.getExecutor();
    eventExecutor.setCorePoolSize(eventThreads + 1);
    eventExecutor.setMaximumPoolSize(eventThreads + 1);
    eventExecutor.setKeepAliveTime(60, TimeUnit.SECONDS);

    socketAcceptor.getDefaultConfig().setThreadModel(threadModel);
    // Add the XMPP codec filter
    socketAcceptor.getFilterChain().addFirst("xmpp", new ProtocolCodecFilter(new XMPPCodecFactory()));
    // Kill sessions whose outgoing queues keep growing and fail to send traffic
    socketAcceptor.getFilterChain().addAfter("xmpp", "outCap", new StalledSessionsFilter());

    try {
        // Listen on a specific network interface if it has been set.
        String interfaceName = JiveGlobals.getXMLProperty("xmpp.socket.network.interface");
        InetAddress bindInterface = null;
        if (interfaceName != null) {
            if (interfaceName.trim().length() > 0) {
                bindInterface = InetAddress.getByName(interfaceName);
            }
        }
        // Start accepting connections
        socketAcceptor.bind(new InetSocketAddress(bindInterface, port), new ClientConnectionHandler());

        List<String> params = new ArrayList<String>();
        params.add(Integer.toString(port));
        Log.info(LocaleUtils.getLocalizedString("startup.plain", params));
    }
    catch (Exception e) {
        System.err.println("Error starting XMPP listener on port " + port + ": " +
                e.getMessage());
        Log.error(LocaleUtils.getLocalizedString("admin.error.socket-setup"), e);
    }
}