import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的package包/类
Client(String host, int port, int size, int count) throws Exception
{
    _count = count;
    _size = size;
    AMQDataBlock block = BasicDeliverTest.getDataBlock(size);

    InetSocketAddress address = new InetSocketAddress(host, port);
    ConnectFuture future = new SocketConnector().connect(address, this);
    future.join();
    _session = future.getSession();

    _start = System.currentTimeMillis();
    for(int i = 0; i < count; i++)
    {
        _session.write(block);
    }
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的package包/类
public NetworkConnection connect(ConnectionSettings settings,
        Receiver<java.nio.ByteBuffer> delegate, SSLContextFactory sslFactory)
{
    int transport = getTransport(settings.getProtocol());
    
    IoConnectorCreator stc;
    switch(transport)
    {
        case TCP:
            stc = new IoConnectorCreator(new SocketConnectorFactory()
            {
                public IoConnector newConnector()
                {
                    return new SocketConnector(1, new QpidThreadExecutor()); // non-blocking connector
                }
            });
            _connection = stc.connect(delegate, settings, sslFactory);
            break;
        case UNKNOWN:
        default:
                throw new TransportException("Unknown protocol: " + settings.getProtocol());
    }

    return _connection;
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的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.SocketConnector; //导入依赖的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.SocketConnector; //导入依赖的package包/类
private void closeConnection() {
    if (connector instanceof SocketConnector) {
        // Change the worker timeout to 0 second to make the I/O thread quit soon when there's no connection to manage.
        // Default worker timeout is 60 sec and therefore the client using MinaProducer cannot terminate the JVM
        // asap but must wait for the timeout to happen, so to speed this up we set the timeout to 0.
        LOG.trace("Setting SocketConnector WorkerTimeout=0 to force MINA stopping its resources faster");
        ((SocketConnector) connector).setWorkerTimeout(0);
    }

    if (session != null) {
        session.close();
    }
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的package包/类
@Test
public void testProducerShutdownTestingWithMock() throws Exception {
    MockEndpoint mock = getMockEndpoint("mock:result");
    mock.expectedBodiesReceived("Hello World");

    // create our mock and record expected behavior = that worker timeout should be set to 0
    SocketConnector mockConnector = createMock(SocketConnector.class);
    mockConnector.setWorkerTimeout(0);
    replay(mockConnector);

    // normal camel code to get a producer
    Endpoint endpoint = context.getEndpoint("mina:tcp://localhost:{{port}}?textline=true&sync=false");
    Exchange exchange = endpoint.createExchange();
    Producer producer = endpoint.createProducer();
    producer.start();

    // set input and execute it
    exchange.getIn().setBody("Hello World");
    producer.process(exchange);

    // insert our mock instead of real MINA IoConnector
    Field field = producer.getClass().getDeclaredField("connector");
    field.setAccessible(true);
    field.set(producer, mockConnector);

    // stop using our mock
    producer.stop();

    verify(mockConnector);

    assertMockEndpointsSatisfied();
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的package包/类
public IoSessionFactory(NioLogger logger,
                        String hosts,
                        JMXReportingThreadPoolExecutor executorService,
                        JMXReportingThreadPoolExecutor reconnectExecutor,
                        NioConfig config,
                        IoHandler ioHandler,
                        IoFutureListener sessionClosedListener,
                        int reconnectInterval,
                        int handshakeResponseTimeout,
                        long sessionRecycleInterval,
                        NetworkAddressResolver addressResolver) {
    this.logger = logger;
    this.reconnectInterval = reconnectInterval;
    this.handshakeResponseTimeout = handshakeResponseTimeout;

    this.hosts = hosts;

    this.nioConfig = config;

    this.socketConnector = new SocketConnector(executorService.getCorePoolSize(), executorService);
    this.socketConnector.setWorkerTimeout(config.getWorkerTimeout());

    this.ioHandler = ioHandler;
    this.sessionClosedListener = sessionClosedListener;

    this.keepRunning = false;
    this.reconnectExecutor = reconnectExecutor;
    sessionRecycler = new SessionRecycler(this, addressResolver, hosts, sessionRecycleInterval);
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的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.SocketConnector; //导入依赖的package包/类
public Fuzz (String host)
{
  executor = newCachedThreadPool ();
  
  // connector
  connector = new SocketConnector (1, executor);
  connector.setWorkerTimeout (0);
  
  remoteAddress = new InetSocketAddress (host, 2917);
  
  random = new Random (hashCode ());
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的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.SocketConnector; //导入依赖的package包/类
private MinaClientFactory() {
  // only one ioConnector,avoid create too many io processor thread
  ioConnector = new SocketConnector(processorCount, Executors.newCachedThreadPool(CONNECTOR_TFACTORY));
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的package包/类
public TairClientFactory() {
	ioConnector = new SocketConnector(processorCount, Executors
			.newCachedThreadPool(CONNECTOR_TFACTORY));
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的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.SocketConnector; //导入依赖的package包/类
public IoSession createClient(NioConfig cfg, IoHandler handler) throws IOException {
    SocketConnector sc = new SocketConnector();
    ConnectFuture cf = sc.connect(new InetSocketAddress("127.0.0.1", 2227), handler, cfg.configureSocketSessionConfig());
    cf.join();
    return cf.getSession();
}
 

import org.apache.mina.transport.socket.nio.SocketConnector; //导入依赖的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 ();
}