View Javadoc

    1   /*
    2    *  Licensed to the Apache Software Foundation (ASF) under one
    3    *  or more contributor license agreements.  See the NOTICE file
    4    *  distributed with this work for additional information
    5    *  regarding copyright ownership.  The ASF licenses this file
    6    *  to you under the Apache License, Version 2.0 (the
    7    *  "License"); you may not use this file except in compliance
    8    *  with the License.  You may obtain a copy of the License at
    9    *
    10   *    http://www.apache.org/licenses/LICENSE-2.0
    11   *
    12   *  Unless required by applicable law or agreed to in writing,
    13   *  software distributed under the License is distributed on an
    14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    15   *  KIND, either express or implied.  See the License for the
    16   *  specific language governing permissions and limitations
    17   *  under the License.
    18   *
    19   */
    20  package org.apache.mina.transport.socket.nio;
    21
    22  import java.io.IOException;
    23  import java.net.ConnectException;
    24  import java.net.InetSocketAddress;
    25  import java.net.SocketAddress;
    26  import java.nio.channels.SelectionKey;
    27  import java.nio.channels.Selector;
    28  import java.nio.channels.SocketChannel;
    29  import java.util.Queue;
    30  import java.util.Set;
    31  import java.util.concurrent.ConcurrentLinkedQueue;
    32  import java.util.concurrent.Executor;
    33  import java.util.concurrent.atomic.AtomicInteger;
    34
    35  import org.apache.mina.common.ConnectFuture;
    36  import org.apache.mina.common.ExceptionMonitor;
    37  import org.apache.mina.common.IoConnector;
    38  import org.apache.mina.common.IoConnectorConfig;
    39  import org.apache.mina.common.IoHandler;
    40  import org.apache.mina.common.IoServiceConfig;
    41  import org.apache.mina.common.support.AbstractIoFilterChain;
    42  import org.apache.mina.common.support.BaseIoConnector;
    43  import org.apache.mina.common.support.DefaultConnectFuture;
    44  import org.apache.mina.util.NamePreservingRunnable;
    45  import org.apache.mina.util.NewThreadExecutor;
    46
    47  /**
    48   * {@link IoConnector} for socket transport (TCP/IP).
    49   *
    50   * @author The Apache Directory Project (mina-dev@directory.apache.org)
    51   * @version $Rev: 389042 $, $Date: 2006-03-27 07:49:41Z $
    52   */
    53  public class SocketConnector extends BaseIoConnector {
    54      private static final AtomicInteger nextId = new AtomicInteger();
    55
    56      private final Object lock = new Object();
    57
    58      private final int id = nextId.getAndIncrement();
    59
    60      private final String threadName = "SocketConnector-" + id;
    61
    62      private SocketConnectorConfig defaultConfig = new SocketConnectorConfig();
    63
    64      private final Queue<ConnectionRequest> connectQueue = new ConcurrentLinkedQueue<ConnectionRequest>();
    65
    66      private final SocketIoProcessor[] ioProcessors;
    67
    68      private final int processorCount;
    69
    70      private final Executor executor;
    71
    72      private volatile Selector selector;
    73
    74      private Worker worker;
    75
    76      private int processorDistributor = 0;
    77
    78      private int workerTimeout = 60; // 1 min.
    79
    80      /**
    81       * Create a connector with a single processing thread using a NewThreadExecutor
    82       */
    83      public SocketConnector() {
    84          this(1, new NewThreadExecutor());
    85      }
    86
    87      /**
    88       * Create a connector with the desired number of processing threads
    89       *
    90       * @param processorCount Number of processing threads
    91       * @param executor Executor to use for launching threads
    92       */
    93      public SocketConnector(int processorCount, Executor executor) {
    94          if (processorCount < 1) {
    95              throw new IllegalArgumentException(
    96                      "Must have at least one processor");
    97          }
    98
    99          this.executor = executor;
    100         this.processorCount = processorCount;
    101         ioProcessors = new SocketIoProcessor[processorCount];
    102
    103         for (int i = 0; i < processorCount; i++) {
    104             ioProcessors[i] = new SocketIoProcessor(
    105                     "SocketConnectorIoProcessor-" + id + "." + i, executor);
    106         }
    107     }
    108
    109     /**
    110      * How many seconds to keep the connection thread alive between connection requests
    111      *
    112      * @return the number of seconds to keep connection thread alive.
    113      *         0 means that the connection thread will terminate immediately
    114      *         when there's no connection to make.
    115      */
    116     public int getWorkerTimeout() {
    117         return workerTimeout;
    118     }
    119
    120     /**
    121      * Set how many seconds the connection worker thread should remain alive once idle before terminating itself.
    122      *
    123      * @param workerTimeout the number of seconds to keep thread alive.
    124      *                      Must be >=0.  If 0 is specified, the connection
    125      *                      worker thread will terminate immediately when
    126      *                      there's no connection to make.
    127      */
    128     public void setWorkerTimeout(int workerTimeout) {
    129         if (workerTimeout < 0) {
    130             throw new IllegalArgumentException("Must be >= 0");
    131         }
    132         this.workerTimeout = workerTimeout;
    133     }
    134
    135     public ConnectFuture connect(SocketAddress address, IoHandler handler,
    136             IoServiceConfig config) {
    137         return connect(address, null, handler, config);
    138     }
    139
    140     public ConnectFuture connect(SocketAddress address,
    141             SocketAddress localAddress, IoHandler handler,
    142             IoServiceConfig config) {
    143         if (address == null)
    144             throw new NullPointerException("address");
    145         if (handler == null)
    146             throw new NullPointerException("handler");
    147
    148         if (!(address instanceof InetSocketAddress))
    149             throw new IllegalArgumentException("Unexpected address type: "
    150                     + address.getClass());
    151
    152         if (localAddress != null
    153                 && !(localAddress instanceof InetSocketAddress))
    154             throw new IllegalArgumentException(
    155                     "Unexpected local address type: " + localAddress.getClass());
    156
    157         if (config == null) {
    158             config = getDefaultConfig();
    159         }
    160
    161         SocketChannel ch = null;
    162         boolean success = false;
    163         try {
    164             ch = SocketChannel.open();
    165             ch.socket().setReuseAddress(true);
    166             if (localAddress != null) {
    167                 ch.socket().bind(localAddress);
    168             }
    169
    170             ch.configureBlocking(false);
    171
    172             if (ch.connect(address)) {
    173                 DefaultConnectFuture future = new DefaultConnectFuture();
    174                 newSession(ch, handler, config, future);
    175                 success = true;
    176                 return future;
    177             }
    178
    179             success = true;
    180         } catch (IOException e) {
    181             return DefaultConnectFuture.newFailedFuture(e);
    182         } finally {
    183             if (!success && ch != null) {
    184                 try {
    185                     ch.close();
    186                 } catch (IOException e) {
    187                     ExceptionMonitor.getInstance().exceptionCaught(e);
    188                 }
    189             }
    190         }
    191
    192         ConnectionRequest request = new ConnectionRequest(ch, handler, config);
    193         synchronized (lock) {
    194             try {
    195                 startupWorker();
    196             } catch (IOException e) {
    197                 try {
    198                     ch.close();
    199                 } catch (IOException e2) {
    200                     ExceptionMonitor.getInstance().exceptionCaught(e2);
    201                 }
    202
    203                 return DefaultConnectFuture.newFailedFuture(e);
    204             }
    205
    206             connectQueue.add(request);
    207             selector.wakeup();
    208         }
    209
    210         return request;
    211     }
    212
    213     public SocketConnectorConfig getDefaultConfig() {
    214         return defaultConfig;
    215     }
    216
    217     /**
    218      * Sets the config this connector will use by default.
    219      *
    220      * @param defaultConfig the default config.
    221      * @throws NullPointerException if the specified value is <code>null</code>.
    222      */
    223     public void setDefaultConfig(SocketConnectorConfig defaultConfig) {
    224         if (defaultConfig == null) {
    225             throw new NullPointerException("defaultConfig");
    226         }
    227         this.defaultConfig = defaultConfig;
    228     }
    229
    230     private void startupWorker() throws IOException {
    231         synchronized (lock) {
    232             if (worker == null) {
    233                 selector = Selector.open();
    234                 worker = new Worker();
    235                 executor.execute(new NamePreservingRunnable(worker, threadName));
    236             }
    237         }
    238     }
    239
    240     private void registerNew() {
    241         if (connectQueue.isEmpty())
    242             return;
    243
    244         Selector selector = this.selector;
    245         for (;;) {
    246             ConnectionRequest req = connectQueue.poll();
    247
    248             if (req == null)
    249                 break;
    250
    251             SocketChannel ch = req.channel;
    252             try {
    253                 ch.register(selector, SelectionKey.OP_CONNECT, req);
    254             } catch (IOException e) {
    255                 req.setException(e);
    256                 try {
    257                     ch.close();
    258                 } catch (IOException e2) {
    259                     ExceptionMonitor.getInstance().exceptionCaught(e2);
    260                 }
    261             }
    262         }
    263     }
    264
    265     private void processSessions(Set<SelectionKey> keys) {
    266         for (SelectionKey key : keys) {
    267             if (!key.isConnectable())
    268                 continue;
    269
    270             SocketChannel ch = (SocketChannel) key.channel();
    271             ConnectionRequest entry = (ConnectionRequest) key.attachment();
    272
    273             boolean success = false;
    274             try {
    275                 if (ch.finishConnect()) {
    276                     key.cancel();
    277                     newSession(ch, entry.handler, entry.config, entry);
    278                 }
    279                 success = true;
    280             } catch (Throwable e) {
    281                 entry.setException(e);
    282             } finally {
    283                 if (!success) {
    284                     key.cancel();
    285                     try {
    286                         ch.close();
    287                     } catch (IOException e) {
    288                         ExceptionMonitor.getInstance().exceptionCaught(e);
    289                     }
    290                 }
    291             }
    292         }
    293
    294         keys.clear();
    295     }
    296
    297     private void processTimedOutSessions(Set<SelectionKey> keys) {
    298         long currentTime = System.currentTimeMillis();
    299
    300         for (SelectionKey key : keys) {
    301             if (!key.isValid())
    302                 continue;
    303
    304             ConnectionRequest entry = (ConnectionRequest) key.attachment();
    305
    306             if (currentTime >= entry.deadline) {
    307                 entry.setException(new ConnectException());
    308                 try {
    309                     key.channel().close();
    310                 } catch (IOException e) {
    311                     ExceptionMonitor.getInstance().exceptionCaught(e);
    312                 } finally {
    313                     key.cancel();
    314                 }
    315             }
    316         }
    317     }
    318
    319     private void newSession(SocketChannel ch, IoHandler handler,
    320             IoServiceConfig config, ConnectFuture connectFuture)
    321             throws IOException {
    322         SocketSessionImpl session = new SocketSessionImpl(this,
    323                 nextProcessor(), getListeners(), config, ch, handler, ch
    324                         .socket().getRemoteSocketAddress());
    325         try {
    326             getFilterChainBuilder().buildFilterChain(session.getFilterChain());
    327             config.getFilterChainBuilder().buildFilterChain(
    328                     session.getFilterChain());
    329             config.getThreadModel().buildFilterChain(session.getFilterChain());
    330         } catch (Throwable e) {
    331             throw (IOException) new IOException("Failed to create a session.")
    332                     .initCause(e);
    333         }
    334
    335         // Set the ConnectFuture of the specified session, which will be
    336         // removed and notified by AbstractIoFilterChain eventually.
    337         session.setAttribute(AbstractIoFilterChain.CONNECT_FUTURE,
    338                 connectFuture);
    339
    340         // Forward the remaining process to the SocketIoProcessor.
    341         session.getIoProcessor().addNew(session);
    342     }
    343
    344     private SocketIoProcessor nextProcessor() {
    345         if (this.processorDistributor == Integer.MAX_VALUE) {
    346             this.processorDistributor = Integer.MAX_VALUE % this.processorCount;
    347         }
    348
    349         return ioProcessors[processorDistributor++ % processorCount];
    350     }
    351
    352     private class Worker implements Runnable {
    353         private long lastActive = System.currentTimeMillis();
    354
    355         public void run() {
    356             Selector selector = SocketConnector.this.selector;
    357             for (;;) {
    358                 try {
    359                     int nKeys = selector.select(1000);
    360
    361                     registerNew();
    362
    363                     if (nKeys > 0) {
    364                         processSessions(selector.selectedKeys());
    365                     }
    366
    367                     processTimedOutSessions(selector.keys());
    368
    369                     if (selector.keys().isEmpty()) {
    370                         if (System.currentTimeMillis() - lastActive > workerTimeout * 1000L) {
    371                             synchronized (lock) {
    372                                 if (selector.keys().isEmpty()
    373                                         && connectQueue.isEmpty()) {
    374                                     worker = null;
    375                                     try {
    376                                         selector.close();
    377                                     } catch (IOException e) {
    378                                         ExceptionMonitor.getInstance()
    379                                                 .exceptionCaught(e);
    380                                     } finally {
    381                                         SocketConnector.this.selector = null;
    382                                     }
    383                                     break;
    384                                 }
    385                             }
    386                         }
    387                     } else {
    388                         lastActive = System.currentTimeMillis();
    389                     }
    390                 } catch (IOException e) {
    391                     ExceptionMonitor.getInstance().exceptionCaught(e);
    392
    393                     try {
    394                         Thread.sleep(1000);
    395                     } catch (InterruptedException e1) {
    396                         ExceptionMonitor.getInstance().exceptionCaught(e1);
    397                     }
    398                 }
    399             }
    400         }
    401     }
    402
    403     private class ConnectionRequest extends DefaultConnectFuture {
    404         private final SocketChannel channel;
    405
    406         private final long deadline;
    407
    408         private final IoHandler handler;
    409
    410         private final IoServiceConfig config;
    411
    412         private ConnectionRequest(SocketChannel channel, IoHandler handler,
    413                 IoServiceConfig config) {
    414             this.channel = channel;
    415             long timeout;
    416             if (config instanceof IoConnectorConfig) {
    417                 timeout = ((IoConnectorConfig) config)
    418                         .getConnectTimeoutMillis();
    419             } else {
    420                 timeout = ((IoConnectorConfig) getDefaultConfig())
    421                         .getConnectTimeoutMillis();
    422             }
    423             this.deadline = System.currentTimeMillis() + timeout;
    424             this.handler = handler;
    425             this.config = config;
    426         }
    427     }
    428 }