import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
/**
 * Create a connection to federation listener.
 */
private static IoSession connectFederation (Router router,
                                            EwafURI uri,
                                            IoHandler listener)
{
  SocketConnector connector = new SocketConnector (1, router.executor ());
  SocketConnectorConfig connectorConfig = new SocketConnectorConfig ();
  
  connector.setWorkerTimeout (0);
  
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout (20);
  
  connectorConfig.getFilterChain ().addLast   
    ("codec", FederationFrameCodec.FILTER);
  
  ConnectFuture future = 
    connector.connect (new InetSocketAddress (uri.host, uri.port),
                       listener, connectorConfig);
  
  future.join ();
  
  return future.getSession ();
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
public SimpleClient (String clientName, String hostname, int port)
  throws IOException
{
  this.clientName = clientName;
  
  SocketConnector connector = new SocketConnector ();

  /* Change the worker timeout to 1 second to make the I/O thread
   * quit soon when there's no connection to manage. */
  connector.setWorkerTimeout (1);
  
  SocketConnectorConfig cfg = new SocketConnectorConfig ();
  cfg.setConnectTimeout (10);
  
  cfg.getFilterChain ().addLast ("codec", ClientFrameCodec.FILTER);
  ConnectFuture future =
    connector.connect (new InetSocketAddress (hostname, port),
                       this, cfg);
                                   
  future.join ();
  clientSession = future.getSession ();
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
protected Client createClient(String targetIP, int targetPort, int connectTimeout, String key) throws Exception {
  if (isDebugEnabled) {
    LOGGER.debug("create connection to :" + targetIP + ":" + targetPort + ",timeout is:" + connectTimeout + ",key is:" + key);
  }
  SocketConnectorConfig cfg = new SocketConnectorConfig();
  cfg.setThreadModel(ThreadModel.MANUAL);
  if (connectTimeout > 1000) {
    cfg.setConnectTimeout((int) connectTimeout / 1000);
  } else {
    cfg.setConnectTimeout(1);
  }
  cfg.getSessionConfig().setTcpNoDelay(Boolean.parseBoolean(System.getProperty("nfs.rpc.tcp.nodelay", "true")));
  cfg.getFilterChain().addLast("objectserialize", new MinaProtocolCodecFilter());
  SocketAddress targetAddress = new InetSocketAddress(targetIP, targetPort);
  MinaClientProcessor processor = new MinaClientProcessor(this, key);
  ConnectFuture connectFuture = ioConnector.connect(targetAddress, null, processor, cfg);
  // wait for connection established
  connectFuture.join();

  IoSession ioSession = connectFuture.getSession();
  if ((ioSession == null) || (!ioSession.isConnected())) {
    String targetUrl = targetIP + ":" + targetPort;
    LOGGER.error("create connection error,targetaddress is " + targetUrl);
    throw new Exception("create connection error,targetaddress is " + targetUrl);
  }
  if (isDebugEnabled) {
    LOGGER.debug(
        "create connection to :" + targetIP + ":" + targetPort + ",timeout is:" + connectTimeout + ",key is:" + key + " successed");
  }
  MinaClient client = new MinaClient(ioSession, key, connectTimeout);
  processor.setClient(client);
  return client;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
private synchronized TairClient createClient(String targetUrl, int connectionTimeout, PacketStreamer pstreamer)
		throws Exception {
	SocketConnectorConfig cfg = new SocketConnectorConfig();
	cfg.setThreadModel(ThreadModel.MANUAL);
	if (connectionTimeout < MIN_CONN_TIMEOUT)
		connectionTimeout = MIN_CONN_TIMEOUT;
	cfg.setConnectTimeout((int) connectionTimeout / 1000);
	cfg.getSessionConfig().setTcpNoDelay(true);
	cfg.getFilterChain().addLast("objectserialize",
			new TairProtocolCodecFilter(pstreamer));
	String address = TairUtil.getHost(targetUrl);
	int port = TairUtil.getPort(targetUrl);
	SocketAddress targetAddress = new InetSocketAddress(address, port);
	TairClientProcessor processor = new TairClientProcessor();
	ConnectFuture connectFuture = ioConnector.connect(targetAddress, null,
			processor, cfg);

	connectFuture.join();
	
	IoSession ioSession = connectFuture.getSession();
	if ((ioSession == null) || (!ioSession.isConnected())) {
		throw new Exception(
				"create tair connection error,targetaddress is "
						+ targetUrl);
	}
	if (LOGGER.isTraceEnabled()) {
		LOGGER.trace("create tair connection success,targetaddress is "
				+ targetUrl);
	}
	TairClient client = new TairClient(this, ioSession,targetUrl);
	processor.setClient(client);
	processor.setFactory(this, targetUrl);
	return client;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
/**
 * Open client socket.
 * <p/>
 * throws Exception on errors
 */
public void open() throws Exception {
    if(logger.isDebugEnabled()) {
        logger.debug("Start logger.");
    }

    connector.setWorkerTimeout(10000);

    // Configure the service.
    SocketConnectorConfig cfg = new SocketConnectorConfig();
    cfg.setConnectTimeout(CONNECT_TIMEOUT);
    cfg.setConnectTimeout(10);

    if(useFixCodec) {
        cfg.getFilterChain().addLast("codec",
            new ProtocolCodecFilter(new ServerProtocolCodecFactory()));
    } else {
        cfg.getFilterChain().addLast("codec", new ProtocolCodecFilter(
            new ObjectSerializationCodecFactory()));
    }

    cfg.getFilterChain().addLast("logger", new LoggingFilter());

    for (int i=0; i < 20; i++) {
        try {
            System.out.println("Try connect.");
            final ConnectFuture future =
            connector.connect(new InetSocketAddress(host, port), this, cfg);

            future.join();
            session = future.getSession();
            return;
        } catch (RuntimeIOException e) {
            Thread.sleep(5000);
        }
    }

    throw new BuildException("Failed connect.");
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
/**
 * Create a connection to federation listener.
 */
private static IoSession connectFederationTLS (Router router,
                                               EwafURI uri,
                                               IoHandler listener)
  throws Exception
{
  SocketConnector connector = new SocketConnector (1, router.executor ());
  SocketConnectorConfig connectorConfig = new SocketConnectorConfig ();
  
  connector.setWorkerTimeout (0);
  
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout (20);
  
  SSLFilter filter = new SSLFilter (defaultSSLContext ());
  
  filter.setUseClientMode (true);
  
  connectorConfig.getFilterChain ().addFirst ("ssl", filter);   
  
  connectorConfig.getFilterChain ().addLast   
    ("codec", FederationFrameCodec.FILTER);
  
  ConnectFuture future = 
    connector.connect (new InetSocketAddress (uri.host, uri.port),
                       listener, connectorConfig);
  
  future.join ();
  
  return future.getSession ();
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
private IoSession connect ()
{
  SocketConnectorConfig connectorConfig = new SocketConnectorConfig ();
  
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout (20);
  
  ConnectFuture future = 
    connector.connect (remoteAddress, new IoHandlerAdapter (),
                       connectorConfig);
  
  future.join ();
  
  return future.getSession ();
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
private static SocketConnectorConfig createTLSConfig () 
  throws Exception
{
  SocketConnectorConfig connectorConfig = createStandardConfig ();
  
  SSLContext sslContext = SSLContext.getInstance ("TLS");
  sslContext.init (null, new TrustManager [] {ACCEPT_ALL_MANAGER}, null);
  
  SSLFilter sslFilter = new SSLFilter (sslContext);
  sslFilter.setUseClientMode (true);
  
  connectorConfig.getFilterChain ().addFirst  ("ssl", sslFilter);
                              
  return connectorConfig;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
private static SocketConnectorConfig createStandardConfig () 
  throws Exception
{
  SocketConnectorConfig connectorConfig = new SocketConnectorConfig ();
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout (10);
  
  connectorConfig.getFilterChain ().addLast 
    ("codec", ClientFrameCodec.FILTER);
  
  return connectorConfig;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
public AcceptorConnectorSetup ()
  throws IOException
{
  executor = newCachedThreadPool ();
  
  // listener
  acceptor = new SocketAcceptor (1, executor);
  acceptorConfig = new SocketAcceptorConfig ();
  
  acceptorConfig.setReuseAddress (true);
  acceptorConfig.setThreadModel (ThreadModel.MANUAL);
  
  DefaultIoFilterChainBuilder filterChainBuilder =
    acceptorConfig.getFilterChain ();

  filterChainBuilder.addLast ("codec", ClientFrameCodec.FILTER);
  
  // connector
  connector = new SocketConnector (1, executor);
  connectorConfig = new SocketConnectorConfig ();
  
  connector.setWorkerTimeout (0);
  
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout (20);
  
  connectorConfig.getFilterChain ().addLast   
    ("codec", ClientFrameCodec.FILTER);
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
protected MinaEndpoint createSocketEndpoint(String uri, MinaConfiguration configuration) {
    boolean minaLogger = configuration.isMinaLogger();
    long timeout = configuration.getTimeout();
    boolean sync = configuration.isSync();
    List<IoFilter> filters = configuration.getFilters();
    final int processorCount = Runtime.getRuntime().availableProcessors() + 1;

    ExecutorService acceptorPool = getCamelContext().getExecutorServiceManager().newCachedThreadPool(this, "MinaSocketAcceptor");
    ExecutorService connectorPool = getCamelContext().getExecutorServiceManager().newCachedThreadPool(this, "MinaSocketConnector");
    ExecutorService workerPool = getCamelContext().getExecutorServiceManager().newCachedThreadPool(this, "MinaThreadPool");

    IoAcceptor acceptor = new SocketAcceptor(processorCount, acceptorPool);
    IoConnector connector = new SocketConnector(processorCount, connectorPool);
    SocketAddress address = new InetSocketAddress(configuration.getHost(), configuration.getPort());

    // connector config
    SocketConnectorConfig connectorConfig = new SocketConnectorConfig();
    // must use manual thread model according to Mina documentation
    connectorConfig.setThreadModel(ThreadModel.MANUAL);
    configureCodecFactory("MinaProducer", connectorConfig, configuration);
    connectorConfig.getFilterChain().addLast("threadPool", new ExecutorFilter(workerPool));
    if (minaLogger) {
        connectorConfig.getFilterChain().addLast("logger", new LoggingFilter());
    }
    appendIoFiltersToChain(filters, connectorConfig.getFilterChain());

    // set connect timeout to mina in seconds
    connectorConfig.setConnectTimeout((int) (timeout / 1000));

    // acceptor connectorConfig
    SocketAcceptorConfig acceptorConfig = new SocketAcceptorConfig();
    // must use manual thread model according to Mina documentation
    acceptorConfig.setThreadModel(ThreadModel.MANUAL);
    configureCodecFactory("MinaConsumer", acceptorConfig, configuration);
    acceptorConfig.setReuseAddress(true);
    acceptorConfig.setDisconnectOnUnbind(true);
    acceptorConfig.getFilterChain().addLast("threadPool", new ExecutorFilter(workerPool));
    if (minaLogger) {
        acceptorConfig.getFilterChain().addLast("logger", new LoggingFilter());
    }
    appendIoFiltersToChain(filters, acceptorConfig.getFilterChain());

    MinaEndpoint endpoint = new MinaEndpoint(uri, this);
    endpoint.setAddress(address);
    endpoint.setAcceptor(acceptor);
    endpoint.setAcceptorConfig(acceptorConfig);
    endpoint.setConnector(connector);
    endpoint.setConnectorConfig(connectorConfig);
    endpoint.setConfiguration(configuration);

    // enlist threads pools in use on endpoint
    endpoint.addThreadPool(acceptorPool);
    endpoint.addThreadPool(connectorPool);
    endpoint.addThreadPool(workerPool);

    // set sync or async mode after endpoint is created
    if (sync) {
        endpoint.setExchangePattern(ExchangePattern.InOut);
    } else {
        endpoint.setExchangePattern(ExchangePattern.InOnly);
    }

    return endpoint;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的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.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
public synchronized SocketConnectorConfig configureSocketSessionConfig() throws IOException {
    SocketConnectorConfig config = new SocketConnectorConfig();
    configureSocketSessionConfig(config.getSessionConfig());
    configureProtocol(config, false);
    return config;
}
 

import org.apache.mina.transport.socket.nio.SocketConnectorConfig; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public Connector (Router router, String serverDomain,
                  EwafURI uri, FederationClass federationClass,
                  Options options) 
  throws IllegalConfigOptionException, IOException
{
  this.router = router;
  this.uri = uri;
  this.serverDomain = serverDomain;
  this.federationClass = federationClass;
  this.options = options;
  this.connector = new SocketConnector (1, router.executor ());
  this.connectorConfig = new SocketConnectorConfig ();
  this.remoteAddress = new InetSocketAddress (uri.host, uri.port);
  
  long requestTimeout = options.getInt ("Federation.Request-Timeout") * 1000;
  long keepaliveInterval = 
    options.getInt ("Federation.Keepalive-Interval") * 1000;

  /* Change the worker timeout to make the I/O thread quit soon
   * when there's no connection to manage. */
  connector.setWorkerTimeout (0);
  
  connectorConfig.setThreadModel (ThreadModel.MANUAL);
  connectorConfig.setConnectTimeout ((int)(requestTimeout / 1000));
  
  DefaultIoFilterChainBuilder filters = new DefaultIoFilterChainBuilder ();
  
  filters.addLast ("codec", FederationFrameCodec.FILTER);
  
  filters.addLast
    ("requestTracker", new RequestTrackingFilter (requestTimeout));
  
  filters.addLast
    ("liveness", new LivenessFilter (keepaliveInterval, requestTimeout));

  Filter<InetAddress> authRequired = 
    (Filter<InetAddress>)options.get ("Federation.Require-Authenticated");
  
  if (uri.isSecure ())
    filters = router.createSecureFilters (filters, authRequired, true);
  else
    filters = router.createStandardFilters (filters, authRequired);
  
  connectorConfig.setFilterChainBuilder (filters);
  
  connect ();
}