##### Coverage Report - com.sun.grizzly.Controller

Classes in this File | Line Coverage | Branch Coverage | Complexity
Controller |
75 % | 272/363
63 % | 121/192
| 0.0;0
Controller$1 |
100 % | 2/2
N/A | N/A
| 0.0;0
Controller$2 |
57 % | 4/7
N/A | N/A
| 0.0;0
Controller$Protocol |
100 % | 1/1
N/A | N/A
| 0.0;0

1 |  |
     /*
2 |  |
      *
3 |  |
      * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
4 |  |
      *
5 |  |
      * Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.
6 |  |
      *
7 |  |
      * The contents of this file are subject to the terms of either the GNU
8 |  |
      * General Public License Version 2 only ("GPL") or the Common Development
9 |  |
      * and Distribution License("CDDL") (collectively, the "License").  You
10 |  |
      * may not use this file except in compliance with the License. You can obtain
11 |  |
      * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
12 |  |
      * or glassfish/bootstrap/legal/LICENSE.txt.  See the License for the specific
13 |  |
      * language governing permissions and limitations under the License.
14 |  |
      *
15 |  |
      * When distributing the software, include this License Header Notice in each
16 |  |
      * file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
17 |  |
      * Sun designates this particular file as subject to the "Classpath" exception
18 |  |
      * as provided by Sun in the GPL Version 2 section of the License file that
19 |  |
      * accompanied this code.  If applicable, add the following below the License
20 |  |
      * Header, with the fields enclosed by brackets [] replaced by your own
21 |  |
      * identifying information: "Portions Copyrighted [year]
22 |  |
      * [name of copyright owner]"
23 |  |
      *
24 |  |
      * Contributor(s):
25 |  |
      *
26 |  |
      * If you wish your version of this file to be governed by only the CDDL or
27 |  |
      * only the GPL Version 2, indicate your decision by adding "[Contributor]
28 |  |
      * elects to include this software in this distribution under the [CDDL or GPL
29 |  |
      * Version 2] license."  If you don't indicate a single choice of license, a
30 |  |
      * recipient has the option to distribute your version of this file under
31 |  |
      * either the CDDL, the GPL Version 2 or to extend the choice of license to
32 |  |
      * its licensees as provided above.  However, if you add GPL Version 2 code
33 |  |
      * and therefore, elected the GPL Version 2 license, then the option applies
34 |  |
      * only if the new code is made subject to such option by the copyright
35 |  |
      * holder.
36 |  |
      *
37 |  |
      */
38 |  |
39 |  |
     package com.sun.grizzly;
40 |  |
41 |  |
     import com.sun.grizzly.util.AttributeHolder;
42 |  |
     import com.sun.grizzly.util.Cloner;
43 |  |
     import com.sun.grizzly.util.ConcurrentLinkedQueuePool;
44 |  |
     import com.sun.grizzly.util.Copyable;
45 |  |
     import com.sun.grizzly.util.State;
46 |  |
     import com.sun.grizzly.util.StateHolder;
47 |  |
     import com.sun.grizzly.util.SupportStateHolder;
48 |  |
     import java.io.IOException;
49 |  |
     import java.nio.channels.ClosedChannelException;
50 |  |
     import java.nio.channels.ClosedSelectorException;
51 |  |
     import java.nio.channels.SelectionKey;
52 |  |
     import java.nio.channels.Selector;
53 |  |
     import java.util.Collection;
54 |  |
     import java.util.HashMap;
55 |  |
     import java.util.Iterator;
56 |  |
     import java.util.Map;
57 |  |
     import java.util.Set;
58 |  |
     import java.util.concurrent.Callable;
59 |  |
     import java.util.concurrent.ConcurrentLinkedQueue;
60 |  |
     import java.util.concurrent.CountDownLatch;
61 |  |
     import java.util.concurrent.atomic.AtomicInteger;
62 |  |
     import java.util.logging.Level;
63 |  |
     import java.util.logging.Logger;
64 |  |
65 |  |
     import static com.sun.grizzly.Context.OpType;
66 |  |
67 |  |
     /**
68 |  |
      * <p>
69 |  |
      * Main entry point when using the Grizzly Framework. A Controller is composed
70 |  |
      * of Handlers, ProtocolChain and Pipeline. All of those components are
71 |  |
      * configurable by client using the Grizzly Framework.
72 |  |
      * </p>
73 |  |
      *
74 |  |
      * <p>
75 |  |
      * A Pipeline is a wrapper around a Thread pool.
76 |  |
      * </p>
77 |  |
      * <p>
78 |  |
      * A ProtocolChain implement the "Chain of Responsibility" pattern (for more info,
79 |  |
      * take a look at the classic "Gang of Four" design patterns book). Towards
80 |  |
      * that end, the Chain API models a computation as a series of "protocol filter"
81 |  |
      * that can be combined into a "protocol chain".
82 |  |
      * </p>
83 |  |
      * <p>
84 |  |
      * An Handler is a interface that can be implemented
85 |  |
      * by implemented by client of the Grizzly Framework to used to help handling
86 |  |
      * NIO operations. The Grizzly Framework define three Handlers:
87 |  |
      * </p>
88 |  |
      * <p><pre><code>
89 |  |
      * (1) SelectorHandler: A SelectorHandler handles all java.nio.channels.Selector
90 |  |
      *                     operations. One or more instance of a Selector are
91 |  |
      *                     handled by SelectorHandler. The logic for processing of
92 |  |
      *                     SelectionKey interest (OP_ACCEPT,OP_READ, etc.) is usually
93 |  |
      *                     defined using an instance of SelectorHandler.
94 |  |
      * (2) SelectionKeyHandler: A SelectionKeyHandler is used to handle the life
95 |  |
      *                          life cycle of a SelectionKey. Operations like canceling,
96 |  |
      *                          registering or closing are handled by SelectionKeyHandler.
97 |  |
      * (3) ProtocolChainInstanceHandler: An ProtocolChainInstanceHandler is where one or several ProtocolChain
98 |  |
      *                      are created and cached. An ProtocolChainInstanceHandler decide if
99 |  |
      *                      a stateless or statefull ProtocolChain needs to be created.
100 |  |
      * </code></pre></p>
101 |  |
      * <p>
102 |  |
      * By default, the Grizzly Framework bundles implementation for TCP
103 |  |
      * and UPD transport. The TCPSelectorHandler is instanciated by default. As an
104 |  |
      * example, supporting the HTTP protocol should only consist of adding the
105 |  |
      * appropriate ProtocolFilter like:
106 |  |
      * </p>
107 |  |
      * <p><pre><code>
108 |  |
      *       Controller sel = new Controller();
109 |  |
      *       sel.setProtocolChainInstanceHandler(new DefaultProtocolChainInstanceHandler(){
110 |  |
      *           public ProtocolChain poll() {
111 |  |
      *               ProtocolChain protocolChain = protocolChains.poll();
112 |  |
      *               if (protocolChain == null){
113 |  |
      *                   protocolChain = new DefaultProtocolChain();
114 |  |
      *                   protocolChain.addFilter(new ReadFilter());
115 |  |
      *                   protocolChain.addFilter(new HTTPParserFilter());
116 |  |
      *               }
117 |  |
      *               return protocolChain;
118 |  |
      *           }
119 |  |
      *       });
120 |  |
      *
121 |  |
      * </code></pre></p>
122 |  |
      * <p>
123 |  |
      * In the example above, a pool of ProtocolChain will be created, and all instance
124 |  |
      * of ProtocolChain will have their instance of ProtocolFilter. Hence the above
125 |  |
      * implementation can be called statefull. A stateless implementation would
126 |  |
      * instead consist of sharing the ProtocolFilter among ProtocolChain:
127 |  |
      * </p>
128 |  |
      * <p><pre><code>
129 |  |
      *       final Controller sel = new Controller();
130 |  |
      *       final ReadFilter readFilter = new ReadFilter();
131 |  |
      *       final LogFilter logFilter = new LogFilter();
132 |  |
      *
133 |  |
      *       sel.setProtocolChainInstanceHandler(new DefaultProtocolChainInstanceHandler(){
134 |  |
      *           public ProtocolChain poll() {
135 |  |
      *               ProtocolChain protocolChain = protocolChains.poll();
136 |  |
      *               if (protocolChain == null){
137 |  |
      *                   protocolChain = new DefaultProtocolChain();
138 |  |
      *                   protocolChain.addFilter(readFilter);
139 |  |
      *                   protocolChain.addFilter(logFilter);
140 |  |
      *               }
141 |  |
      *               return protocolChain;
142 |  |
      *           }
143 |  |
      *       });
144 |  |
      * </code></pre></p>
145 |  |
      * @author Jeanfrancois Arcand
146 |  |
      */
147 |  |
     public class Controller implements Runnable, Lifecycle, Copyable,
148 |  |
             ConnectorHandlerPool, AttributeHolder, SupportStateHolder<State> {
149 |  |
150 | 5 |
         public enum Protocol { UDP, TCP , TLS, CUSTOM }
151 |  |
152 |  |
153 |  |
         /**
154 |  |
          * A cached list of Context. Context are by default stateless.
155 |  |
          */
156 |  |
         private ConcurrentLinkedQueuePool<Context> contexts;
157 |  |
158 |  |
159 |  |
         /**
160 |  |
          * The ProtocolChainInstanceHandler used by this instance. If not set, and instance
161 |  |
          * of the DefaultInstanceHandler will be created.
162 |  |
          */
163 |  |
         protected ProtocolChainInstanceHandler instanceHandler;
164 |  |
165 |  |
166 |  |
         /**
167 |  |
          * The SelectionKey Handler used by this instance. If not set, and instance
168 |  |
          * of the DefaultSelectionKeyHandler will be created.
169 |  |
          */
170 |  |
         protected SelectionKeyHandler selectionKeyHandler;
171 |  |
172 |  |
173 |  |
         /**
174 |  |
          * The SelectorHandler, which will manage connection accept,
175 |  |
          * if readThreadsCount > 0 and spread connection processing between
176 |  |
          * different read threads
177 |  |
          */
178 | 85 |
         protected ComplexSelectorHandler multiReadThreadSelectorHandler = null;
179 |  |
180 |  |
181 |  |
         /**
182 |  |
          * The ConnectorHandlerPool, which is responsible for creating/caching
183 |  |
          * ConnectorHandler instances.
184 |  |
          */
185 | 85 |
         protected ConnectorHandlerPool connectorHandlerPool = null;
186 |  |
187 |  |
188 |  |
         /**
189 |  |
          * The set of {@link SelectorHandler}s used by this instance. If not set, the instance
190 |  |
          * of the TCPSelectorHandler will be added by default.
191 |  |
          */
192 |  |
         protected ConcurrentLinkedQueue<SelectorHandler> selectorHandlers;
193 |  |
194 |  |
195 |  |
         /**
196 |  |
          * Current {@link Controller} state
197 |  |
          */
198 |  |
         protected StateHolder<State> stateHolder;
199 |  |
200 |  |
201 |  |
         /**
202 |  |
          * The number of read threads
203 |  |
          */
204 | 85 |
         protected int readThreadsCount = 0;
205 |  |
206 |  |
207 |  |
         /**
208 |  |
          * The array of {@link Controller}s to be used for reading
209 |  |
          */
210 |  |
         protected ReadController[] readThreadControllers;
211 |  |
212 |  |
213 |  |
         /**
214 |  |
          * Default Logger.
215 |  |
          */
216 | 1 |
         private static Logger logger = Logger.getLogger("grizzly");
217 |  |
218 |  |
219 |  |
         /**
220 |  |
          * Default Thread Pool (called Pipeline).If not set, and instance
221 |  |
          * of the DefaultPipeline will be created.
222 |  |
          */
223 |  |
         private Pipeline<Callable> pipeline;
224 |  |
225 |  |
226 |  |
         /**
227 |  |
          * Collection of {@link Controller} state listeners, which
228 |  |
          * will are notified on {@link Controller} state change.
229 |  |
          */
230 | 85 |
         protected Collection<ControllerStateListener> stateListeners =
231 |  |
                 new ConcurrentLinkedQueue<ControllerStateListener>();
232 |  |
233 |  |
234 |  |
         /**
235 |  |
          * Internal countdown counter of {@link SelectorHandler}s, which
236 |  |
          * are ready to process
237 |  |
          */
238 |  |
         protected AtomicInteger readySelectorHandlerCounter;
239 |  |
240 |  |
         /**
241 |  |
          * Internal countdown counter of {@link SelectorHandler}s, which stopped
242 |  |
          */
243 |  |
         protected AtomicInteger stoppedSelectorHandlerCounter;
244 |  |
245 |  |
246 |  |
         /**
247 |  |
          * <tt>true</tt> if OP_READ and OP_WRITE can be handled concurrently.
248 |  |
          */
249 | 85 |
         private boolean handleReadWriteConcurrently = true;
250 |  |
251 |  |
252 |  |
         /**
253 |  |
          * Attributes, associated with the {@link Controller} instance
254 |  |
          */
255 |  |
         protected Map<String, Object> attributes;
256 |  |
257 |  |
258 |  |
         /**
259 |  |
          * The current Controller instance.
260 |  |
          *
261 |  |
          */
262 | 1 |
         private final static ConcurrentLinkedQueue<Controller> controllers =
263 |  |
                 new ConcurrentLinkedQueue<Controller>();
264 |  |
265 |  |
266 |  |
         // -------------------------------------------------------------------- //
267 |  |
268 |  |
         /**
269 |  |
          * Controller constructor
270 |  |
          */
271 | 85 |
         public Controller() {
272 | 85 |
             contexts = new ConcurrentLinkedQueuePool<Context>() {
273 |  |
                 @Override
274 |  |
                 public Context newInstance() {
275 | 261 |
                     return new Context();
276 |  |
                 }
277 |  |
             };
278 |  |
279 | 85 |
             stateHolder = new StateHolder<State>(true);
280 | 85 |
             initializeDefaults();
281 | 85 |
         }
282 |  |
283 |  |
284 |  |
         /**
285 |  |
          * This method initializes this Controller's default Pipeline,
286 |  |
          * default ProtocolChainInstanceHandler, default SelectorHandler(s)
287 |  |
          * and default ConnectorHandlerPool.  These defaults can be overridden
288 |  |
          * after this Controller constructor is called and before calling
289 |  |
          * Controller.start() using this Controller's mutator methods to
290 |  |
          * set a different Pipeline, ProtocolChainInstanceHandler,
291 |  |
          * SelectorHandler(s) or ConnectorHandlerPool.
292 |  |
          */
293 |  |
         private void initializeDefaults() {
294 | 85 |
             if (pipeline == null) {
295 | 85 |
                 pipeline = new DefaultPipeline();
296 |  |
             }
297 | 85 |
             if (instanceHandler == null) {
298 | 85 |
                 instanceHandler = new DefaultProtocolChainInstanceHandler();
299 |  |
             }
300 | 85 |
             if (selectorHandlers == null){
301 | 85 |
                 selectorHandlers = new ConcurrentLinkedQueue<SelectorHandler>();
302 |  |
             }
303 | 85 |
             if (connectorHandlerPool == null) {
304 | 85 |
                 connectorHandlerPool = new DefaultConnectorHandlerPool(this);
305 |  |
             }
306 | 85 |
             controllers.add(this);
307 | 85 |
         }
308 |  |
309 |  |
310 |  |
         /**
311 |  |
          * This method handle the processing of all Selector's interest op
312 |  |
          * (OP_ACCEPT,OP_READ,OP_WRITE,OP_CONNECT) by delegating to its Handler.
313 |  |
          * By default, all java.nio.channels.Selector operations are implemented
314 |  |
          * using SelectorHandler. All SelectionKey operations are implemented by
315 |  |
          * SelectionKeyHandler. Finally, ProtocolChain creation/re-use are implemented
316 |  |
          * by InstanceHandler.
317 |  |
          * @param selectorHandler - the {@link SelectorHandler}
318 |  |
          */
319 |  |
         protected void doSelect(SelectorHandler selectorHandler){
320 | 266289 |
             SelectionKey key = null;
321 |  |
             Set<SelectionKey> readyKeys;
322 |  |
             Iterator<SelectionKey> iterator;
323 |  |
             int selectorState;
324 | 266289 |
             boolean delegateToWorkerThread = false;
325 | 266289 |
             Context serverCtx = contexts.poll();
326 | 266289 |
             serverCtx.setController(this);
327 |  |
328 | 266289 |
             serverCtx.setSelectorHandler(selectorHandler);
329 |  |
330 |  |
             try {
331 | 266289 |
                 selectorState = 0;
332 |  |
333 |  |
                 // Set the SelectionKeyHandler only if the SelectorHandler doesn't
334 |  |
                 // define one.
335 | 266289 |
                 if (selectorHandler.getSelectionKeyHandler() == null){
336 | 123 |
                     if (logger.isLoggable(Level.FINE)) {
337 | 0 |
                         logger.log(Level.FINE, "Set DefaultSelectionKeyHandler to SelectorHandler: " + selectorHandler);
338 |  |
                     }
339 |  |
340 |  |
341 | 123 |
                     SelectionKeyHandler assgnSelectionKeyHandler = null;
342 |  |
343 | 123 |
                     if (selectorHandler.getPreferredSelectionKeyHandler() != null) {
344 | 123 |
                         Class<? extends SelectionKeyHandler> keyHandlerClass =
345 |  |
                                     selectorHandler.getPreferredSelectionKeyHandler();
346 |  |
                         try {
347 | 123 |
                             assgnSelectionKeyHandler = keyHandlerClass.newInstance();
348 | 123 |
                             assgnSelectionKeyHandler.setSelectorHandler(selectorHandler);
349 | 0 |
                         } catch (Exception e) {
350 | 0 |
                             if (logger.isLoggable(Level.WARNING)) {
351 | 0 |
                                 logger.log(Level.WARNING,
352 |  |
                                         "Exception initializing preffered SelectionKeyHandler '" +
353 |  |
                                         keyHandlerClass + "' for the SelectorHandler '" +
354 |  |
                                         selectorHandler + "'");
355 |  |
                             }
356 | 123 |
                         }
357 |  |
                     }
358 |  |
359 | 123 |
                     if (assgnSelectionKeyHandler == null) {
360 | 0 |
                         assgnSelectionKeyHandler =
361 |  |
                                 new DefaultSelectionKeyHandler(selectorHandler);
362 |  |
                     }
363 |  |
364 | 123 |
                     selectorHandler.setSelectionKeyHandler(assgnSelectionKeyHandler);
365 |  |
                 }
366 |  |
367 | 266289 |
                 selectorHandler.preSelect(serverCtx);
368 |  |
369 | 266289 |
                 readyKeys = selectorHandler.select(serverCtx);
370 | 266289 |
                 selectorState = readyKeys.size();
371 | 266289 |
                 if (stateHolder.getState(false) == State.STARTED &&
372 |  |
                         selectorHandler.getStateHolder().getState(false) == State.STARTED &&
373 |  |
                         selectorState != 0) {
374 | 138808 |
                     iterator = readyKeys.iterator();
375 | 300214 |
                     while (iterator.hasNext()) {
376 | 161406 |
                         key = iterator.next();
377 | 161406 |
                         iterator.remove();
378 | 161406 |
                         OpType opType = null;
379 | 161406 |
                         boolean skipOpWrite = false;
380 | 161406 |
                         delegateToWorkerThread = false;
381 | 161406 |
                         if (key.isValid()) {
382 | 161406 |
                             if ((key.readyOps() & SelectionKey.OP_ACCEPT)
383 |  |
                                     == SelectionKey.OP_ACCEPT){
384 | 199 |
                                 if (readThreadsCount > 0 &&
385 |  |
                                         multiReadThreadSelectorHandler.supportsProtocol(selectorHandler.protocol())) {
386 | 19 |
                                     if (logger.isLoggable(Level.FINE)) {
387 | 0 |
                                         logger.log(Level.FINE, "OP_ACCEPT on " + key + " passed to multi readthread handler");
388 |  |
                                     }
389 | 19 |
                                     delegateToWorkerThread = multiReadThreadSelectorHandler.
390 |  |
                                             onAcceptInterest(key, serverCtx);
391 |  |
                                 } else {
392 | 180 |
                                     if (logger.isLoggable(Level.FINE)) {
393 | 0 |
                                         logger.log(Level.FINE, "OP_ACCEPT on " + key);
394 |  |
                                     }
395 |  |
396 | 180 |
                                     delegateToWorkerThread = selectorHandler.
397 |  |
                                             onAcceptInterest(key, serverCtx);
398 |  |
                                 }
399 | 180 |
                                 continue;
400 |  |
                             }
401 |  |
402 | 161207 |
                             if ((key.readyOps() & SelectionKey.OP_CONNECT)
403 |  |
                                     == SelectionKey.OP_CONNECT) {
404 | 174 |
                                 if (logger.isLoggable(Level.FINE)) {
405 | 0 |
                                     logger.log(Level.FINE, "OP_CONNECT on " + key);
406 |  |
                                 }
407 |  |
408 | 174 |
                                 delegateToWorkerThread = selectorHandler.
409 |  |
                                         onConnectInterest(key, serverCtx);
410 | 174 |
                                 continue;
411 |  |
                             }
412 |  |
413 |  |
                             // OP_READ will always be processed first, then
414 |  |
                             // based on the handleReadWriteConcurrently, the OP_WRITE
415 |  |
                             // might be processed just after or during the next
416 |  |
                             // Selector.select() invocation.
417 | 161033 |
                             if ((key.readyOps() & SelectionKey.OP_READ)
418 |  |
                                     == SelectionKey.OP_READ) {
419 | 158935 |
                                 if (logger.isLoggable(Level.FINE)) {
420 | 0 |
                                     logger.log(Level.FINE, "OP_READ on " + key);
421 |  |
                                 }
422 | 158935 |
                                 delegateToWorkerThread = selectorHandler.
423 |  |
                                         onReadInterest(key,serverCtx);
424 |  |
425 | 158935 |
                                 if (delegateToWorkerThread) {
426 | 90799 |
                                     opType = OpType.OP_READ;
427 |  |
                                 }
428 |  |
429 | 158935 |
                                 if (!handleReadWriteConcurrently){
430 | 137 |
                                     skipOpWrite = true;
431 |  |
                                 }
432 |  |
                             }
433 |  |
434 |  |
                             // The OP_READ processing might have closed the
435 |  |
                             // Selection, hence we must make sure the
436 |  |
                             // SelectionKey is still valid.
437 | 161033 |
                             if (!skipOpWrite && key.isValid()
438 |  |
                                     && (key.readyOps() & SelectionKey.OP_WRITE)
439 |  |
                                     == SelectionKey.OP_WRITE) {
440 | 2241 |
                                 if (logger.isLoggable(Level.FINE)) {
441 | 0 |
                                     logger.log(Level.FINE, "OP_WRITE on " + key);
442 |  |
                                 }
443 | 2241 |
                                 boolean opWriteDelegateToWorkerThread = selectorHandler.
444 |  |
                                         onWriteInterest(key,serverCtx);
445 | 2241 |
                                 delegateToWorkerThread |= opWriteDelegateToWorkerThread;
446 | 2241 |
                                 if (opWriteDelegateToWorkerThread) {
447 | 16 |
                                     if (opType == OpType.OP_READ) {
448 | 14 |
                                         opType = OpType.OP_READ_WRITE;
449 |  |
                                     } else {
450 | 2 |
                                         opType = OpType.OP_WRITE;
451 |  |
                                     }
452 |  |
                                 }
453 |  |
                             }
454 |  |
455 | 161033 |
                             if (delegateToWorkerThread){
456 | 90801 |
                                 Context ctx = pollContext(key, opType);
457 | 90801 |
                                 configureContext(ctx,selectorHandler);
458 | 90801 |
                                 ctx.execute(ProtocolChainContextTask.poll());
459 | 90801 |
                             }
460 |  |
                         } else {
461 | 0 |
                             selectorHandler.getSelectionKeyHandler().cancel(key);
462 |  |
                         }
463 | 161033 |
                     }
464 |  |
                 }
465 |  |
466 | 266289 |
                 delegateToWorkerThread = false;
467 | 266289 |
                 selectorHandler.postSelect(serverCtx);
468 | 266289 |
                 contexts.offer(serverCtx);
469 | 0 |
             } catch (ClosedSelectorException e) {
470 |  |
                 // TODO: This could indicate that the Controller is
471 |  |
                 //       shutting down. Hence, we need to handle this Exception
472 |  |
                 //       appropriately. Perhaps check the state before logging
473 |  |
                 //       what's happening ?
474 | 0 |
                 if (stateHolder.getState() == State.STARTED &&
475 |  |
                         selectorHandler.getStateHolder().getState() == State.STARTED) {
476 | 0 |
                     logger.log(Level.SEVERE, "Selector was unexpectedly closed.");
477 | 0 |
                     notifyException(e);
478 |  |
                 } else {
479 | 0 |
                     logger.log(Level.FINE, "doSelect Selector closed");
480 |  |
                 }
481 | 0 |
             } catch (ClosedChannelException e) {
482 |  |
                 // Don't use stateLock. This case is not strict
483 | 0 |
                 if (stateHolder.getState() == State.STARTED &&
484 |  |
                         selectorHandler.getStateHolder().getState() == State.STARTED) {
485 | 0 |
                     logger.log(Level.WARNING, "Channel was unexpectedly closed");
486 | 0 |
                     if (key != null){
487 | 0 |
                         selectorHandler.getSelectionKeyHandler().cancel(key);
488 |  |
                     }
489 |  |
490 | 0 |
                     notifyException(e);
491 |  |
                 }
492 | 0 |
             } catch (Throwable t) {
493 |  |
                 try{
494 | 0 |
                     logger.log(Level.SEVERE,"doSelect exception",t);
495 | 0 |
                     if (key != null){
496 | 0 |
                         selectorHandler.getSelectionKeyHandler().cancel(key);
497 |  |
                     }
498 |  |
499 | 0 |
                     notifyException(t);
500 | 0 |
                 } catch (Throwable t2){
501 |  |
                     // An unexpected exception occured, most probably caused by
502 |  |
                     // a bad logger. Since logger can be externally configurable,
503 |  |
                     // just output the exception on the screen and continue the
504 |  |
                     // normal execution.
505 | 0 |
                     t2.printStackTrace();
506 | 0 |
                 }
507 | 266289 |
             }
508 | 266289 |
         }
509 |  |
510 |  |
511 |  |
         /**
512 |  |
          * Register a SelectionKey.
513 |  |
          * @param key <tt>SelectionKey</tt> to register
514 |  |
          */
515 |  |
         public void registerKey(SelectionKey key){
516 | 0 |
             registerKey(key,SelectionKey.OP_READ);
517 | 0 |
         }
518 |  |
519 |  |
520 |  |
         /**
521 |  |
          * Register a SelectionKey on the first SelectorHandler that was added
522 |  |
          * using the addSelectorHandler().
523 |  |
          * @param key <tt>SelectionKey</tt> to register
524 |  |
          * @param ops - the interest op to register
525 |  |
          */
526 |  |
         public void registerKey(SelectionKey key, int ops){
527 | 0 |
             registerKey(key, ops, selectorHandlers.peek().protocol());
528 | 0 |
         }
529 |  |
530 |  |
531 |  |
         /**
532 |  |
          * Register a SelectionKey.
533 |  |
          * @param key <tt>SelectionKey</tt> to register
534 |  |
          * @param ops - the interest op to register
535 |  |
          * @param protocol specified protocol SelectorHandler key should be registered on
536 |  |
          */
537 |  |
         public void registerKey(SelectionKey key, int ops, Protocol protocol){
538 | 21 |
             if (stateHolder.getState() == State.STOPPED) {
539 | 0 |
                 return;
540 |  |
             }
541 |  |
542 | 21 |
             getSelectorHandler(protocol).register(key,ops);
543 | 21 |
         }
544 |  |
545 |  |
546 |  |
         /**
547 |  |
          * Cancel a SelectionKey
548 |  |
          * @param key <tt>SelectionKey</tt> to cancel
549 |  |
          * @deprecated
550 |  |
          */
551 |  |
         public void cancelKey(SelectionKey key){
552 | 0 |
             if (stateHolder.getState() == State.STOPPED) {
553 | 0 |
                 return;
554 |  |
             }
555 |  |
556 | 0 |
             SelectorHandler selectorHandler = getSelectorHandler(key.selector());
557 | 0 |
             if (selectorHandler != null) {
558 | 0 |
                 selectorHandler.getSelectionKeyHandler().cancel(key);
559 |  |
             } else {
560 | 0 |
                 throw new IllegalStateException("SelectionKey is not associated " +
561 |  |
                         "with known SelectorHandler");
562 |  |
             }
563 | 0 |
         }
564 |  |
565 |  |
         /**
566 |  |
          * Get an instance of a {@link Context}
567 |  |
          * @param key {@link SelectionKey}
568 |  |
          * @return {@link Context}
569 |  |
          */
570 |  |
         /* package */ public Context pollContext(SelectionKey key) {
571 | 12 |
             return pollContext(key,null);
572 |  |
         }
573 |  |
574 |  |
575 |  |
         /**
576 |  |
          * Get an instance of a {@link Context}
577 |  |
          * @param key {@link SelectionKey}
578 |  |
          * @param opType the current SelectionKey op.
579 |  |
          * @return {@link Context}
580 |  |
          */
581 |  |
         /* package */ public Context pollContext(SelectionKey key, OpType opType) {
582 | 161379 |
             Context ctx = contexts.poll();
583 | 161379 |
             ctx.setController(this);
584 | 161379 |
             ctx.setSelectionKey(key);
585 | 161379 |
             if (opType != null) {
586 | 161367 |
                 ctx.setCurrentOpType(opType);
587 |  |
             } else {
588 | 12 |
                 ctx.configureOpType(key);
589 |  |
             }
590 |  |
591 | 161379 |
             if (logger.isLoggable(Level.FINE)) {
592 | 0 |
                 logger.log(Level.FINE, "pollContext(..) Context : "+ctx);
593 |  |
            }
594 |  |
595 | 161379 |
             return ctx;
596 |  |
         }
597 |  |
598 |  |
         /* package */ public void configureContext(Context ctx,SelectorHandler selectorHandler){
599 | 90804 |
             ctx.setSelectorHandler(selectorHandler);
600 | 90804 |
             ctx.setPipeline(selectorHandler.pipeline());
601 | 90804 |
             ctx.setAsyncQueueReader(selectorHandler.getAsyncQueueReader());
602 | 90804 |
             ctx.setAsyncQueueWriter(selectorHandler.getAsyncQueueWriter());
603 | 90804 |
         }
604 |  |
605 |  |
         /**
606 |  |
          * Return a {@link Context} to its pool if it is not shared.
607 |  |
          *
608 |  |
          * @param ctx - the {@link Context}
609 |  |
          */
610 |  |
         public void returnContext(Context ctx){
611 | 161370 |
             if(ctx.decrementRefCount()>0) {
612 | 0 |
                 return;
613 |  |
             }
614 | 161369 |
            if (logger.isLoggable(Level.FINE)) {
615 | 0 |
                 logger.log(Level.FINE, "returnContext() Context : "+ctx);
616 |  |
             }
617 | 161370 |
             ctx.recycle();
618 | 161370 |
             contexts.offer(ctx);
619 | 161370 |
         }
620 |  |
621 |  |
622 |  |
         /**
623 |  |
          * Return the current <code>Logger</code> used by this Controller.
624 |  |
          */
625 |  |
         public static Logger logger() {
626 | 1256513 |
             return logger;
627 |  |
         }
628 |  |
629 |  |
630 |  |
         /**
631 |  |
          * Set the Logger single instance to use.
632 |  |
          */
633 |  |
         public static void setLogger(Logger l){
634 | 0 |
             logger = l;
635 | 0 |
         }
636 |  |
637 |  |
         // ------------------------------------------------------ Handlers ------//
638 |  |
639 |  |
640 |  |
         /**
641 |  |
          * Set the {@link ProtocolChainInstanceHandler} to use for
642 |  |
          * creating instance of {@link ProtocolChain}.
643 |  |
          */
644 |  |
         public void setProtocolChainInstanceHandler(ProtocolChainInstanceHandler
645 |  |
                 instanceHandler){
646 | 39 |
             this.instanceHandler = instanceHandler;
647 | 39 |
         }
648 |  |
649 |  |
         /**
650 |  |
          * Return the {@link ProtocolChainInstanceHandler}
651 |  |
          */
652 |  |
         public ProtocolChainInstanceHandler getProtocolChainInstanceHandler(){
653 | 322740 |
             return instanceHandler;
654 |  |
         }
655 |  |
656 |  |
657 |  |
         /**
658 |  |
          * @deprecated
659 |  |
          * Set the {@link SelectionKeyHandler} to use for managing the life
660 |  |
          * cycle of SelectionKey.
661 |  |
          * Method is deprecated. Use SelectorHandler.setSelectionKeyHandler() instead
662 |  |
          */
663 |  |
         public void setSelectionKeyHandler(SelectionKeyHandler selectionKeyHandler){
664 | 0 |
             this.selectionKeyHandler = selectionKeyHandler;
665 | 0 |
         }
666 |  |
667 |  |
668 |  |
         /**
669 |  |
          * @deprecated
670 |  |
          * Return the {@link SelectionKeyHandler}
671 |  |
          * Method is deprecated. Use SelectorHandler.getSelectionKeyHandler() instead
672 |  |
          */
673 |  |
         public SelectionKeyHandler getSelectionKeyHandler(){
674 | 0 |
             return selectionKeyHandler;
675 |  |
         }
676 |  |
677 |  |
678 |  |
         /**
679 |  |
          * Add a {@link SelectorHandler}
680 |  |
          * @param selectorHandler - the {@link SelectorHandler}
681 |  |
          */
682 |  |
         public void addSelectorHandler(SelectorHandler selectorHandler) {
683 | 134 |
             selectorHandlers.add(selectorHandler);
684 | 134 |
             if (stateHolder.getState(false) != null &&
685 |  |
                     !State.STOPPED.equals(stateHolder.getState())) {
686 | 48 |
                 addSelectorHandlerOnReadControllers(selectorHandler);
687 | 48 |
                 if (readySelectorHandlerCounter != null) {
688 | 48 |
                     readySelectorHandlerCounter.incrementAndGet();
689 |  |
                 }
690 | 48 |
                 if (stoppedSelectorHandlerCounter != null) {
691 | 48 |
                     stoppedSelectorHandlerCounter.incrementAndGet();
692 |  |
                 }
693 | 48 |
                 startSelectorHandlerRunner(selectorHandler, true);
694 |  |
             }
695 | 134 |
         }
696 |  |
697 |  |
698 |  |
         /**
699 |  |
          * Set the first {@link SelectorHandler}
700 |  |
          * @param selectorHandler - the {@link SelectorHandler}
701 |  |
          */
702 |  |
         public void setSelectorHandler(SelectorHandler selectorHandler){
703 | 68 |
             addSelectorHandler(selectorHandler);
704 | 68 |
         }
705 |  |
706 |  |
707 |  |
         /**
708 |  |
          * Return the {@link SelectorHandler} associated with the protocol.
709 |  |
          * @param protocol - the {@link Controller.Protocol}
710 |  |
          * @return {@link SelectorHandler}
711 |  |
          */
712 |  |
         public SelectorHandler getSelectorHandler(Protocol protocol){
713 | 583 |
             for (SelectorHandler selectorHandler: selectorHandlers){
714 | 583 |
                 if (selectorHandler.protocol() == protocol){
715 | 583 |
                     return selectorHandler;
716 |  |
                 }
717 |  |
             }
718 | 0 |
             return null;
719 |  |
         }
720 |  |
721 |  |
         /**
722 |  |
          * Return the {@link SelectorHandler} associated
723 |  |
          * with the {@link Selector}.
724 |  |
          * @param selector - the {@link Selector}
725 |  |
          * @return {@link SelectorHandler}
726 |  |
          */
727 |  |
         public SelectorHandler getSelectorHandler(Selector selector){
728 | 0 |
             for (SelectorHandler selectorHandler: selectorHandlers){
729 | 0 |
                 if (selectorHandler.getSelector() == selector){
730 | 0 |
                     return selectorHandler;
731 |  |
                 }
732 |  |
             }
733 | 0 |
             return null;
734 |  |
         }
735 |  |
736 |  |
         /**
737 |  |
          * Return the list {@link SelectorHandler}
738 |  |
          * @return {@link ConcurrentLinkedQueue}
739 |  |
          */
740 |  |
         public ConcurrentLinkedQueue getSelectorHandlers(){
741 | 7487 |
             return selectorHandlers;
742 |  |
         }
743 |  |
744 |  |
745 |  |
         /**
746 |  |
          * Shuts down {@link SelectorHandler} and removes it from this
747 |  |
          * {@link Controller} list
748 |  |
          * @param {@link SelectorHandler} to remove
749 |  |
          */
750 |  |
         public void removeSelectorHandler(SelectorHandler selectorHandler) {
751 | 1 |
             if (selectorHandlers.remove(selectorHandler)) {
752 | 1 |
                 removeSelectorHandlerOnReadControllers(selectorHandler);
753 | 1 |
                 selectorHandler.shutdown();
754 |  |
             }
755 | 1 |
         }
756 |  |
757 |  |
758 |  |
         /**
759 |  |
          * Return the {@link Pipeline} (Thread Pool) used by this Controller.
760 |  |
          */
761 |  |
         public Pipeline getPipeline() {
762 | 90877 |
             return pipeline;
763 |  |
         }
764 |  |
765 |  |
766 |  |
         /**
767 |  |
          * Set the {@link Pipeline} (Thread Pool).
768 |  |
          */
769 |  |
         public void setPipeline(Pipeline<Callable> pipeline) {
770 | 34 |
             this.pipeline = pipeline;
771 | 34 |
         }
772 |  |
773 |  |
774 |  |
         /**
775 |  |
          * Return the number of Reader threads count.
776 |  |
          */
777 |  |
         public int getReadThreadsCount() {
778 | 0 |
             return readThreadsCount;
779 |  |
         }
780 |  |
781 |  |
782 |  |
         /**
783 |  |
          * Set the number of Reader threads count.
784 |  |
          */
785 |  |
         public void setReadThreadsCount(int readThreadsCount) {
786 | 21 |
             this.readThreadsCount = readThreadsCount;
787 | 21 |
         }
788 |  |
789 |  |
790 |  |
         /**
791 |  |
          * Return the <code>ConnectorHandlerPool</code> used.
792 |  |
          */
793 |  |
         public ConnectorHandlerPool getConnectorHandlerPool() {
794 | 0 |
             return connectorHandlerPool;
795 |  |
         }
796 |  |
797 |  |
798 |  |
         /**
799 |  |
          * Set the <code>ConnectorHandlerPool</code> used.
800 |  |
          */
801 |  |
         public void setConnectorHandlerPool(ConnectorHandlerPool connectorHandlerPool) {
802 | 1 |
             this.connectorHandlerPool = connectorHandlerPool;
803 | 1 |
         }
804 |  |
805 |  |
         // ------------------------------------------------------ Runnable -------//
806 |  |
807 |  |
808 |  |
         /**
809 |  |
          * Execute this Controller.
810 |  |
          */
811 |  |
         public void run() {
812 |  |
             try{
813 | 84 |
                 start();
814 | 0 |
             } catch(IOException e){
815 | 0 |
                 notifyException(e);
816 | 0 |
                 throw new RuntimeException(e.getCause());
817 | 84 |
             }
818 | 84 |
         }
819 |  |
820 |  |
         // -------------------------------------------------------- Copyable ----//
821 |  |
822 |  |
823 |  |
         /**
824 |  |
          * Copy this Controller state to another instance of a Controller.
825 |  |
          */
826 |  |
         public void copyTo(Copyable copy) {
827 | 17 |
             Controller copyController = (Controller) copy;
828 | 17 |
             copyController.contexts = contexts;
829 | 17 |
             copyController.attributes = attributes;
830 | 17 |
             copyController.instanceHandler = instanceHandler;
831 | 17 |
             copyController.pipeline = pipeline;
832 | 17 |
             copyController.readThreadControllers = readThreadControllers;
833 | 17 |
             copyController.readThreadsCount = readThreadsCount;
834 | 17 |
             copyController.selectionKeyHandler = selectionKeyHandler;
835 | 17 |
             copyController.stateHolder = stateHolder;
836 | 17 |
         }
837 |  |
838 |  |
         // -------------------------------------------------------- Lifecycle ----//
839 |  |
840 |  |
         /**
841 |  |
          * Add controller state listener
842 |  |
          */
843 |  |
         public void addStateListener(ControllerStateListener stateListener) {
844 | 139 |
             stateListeners.add(stateListener);
845 | 139 |
         }
846 |  |
847 |  |
         /**
848 |  |
          * Remove controller state listener
849 |  |
          */
850 |  |
         public void removeStateListener(ControllerStateListener stateListener) {
851 | 69 |
             stateListeners.remove(stateListener);
852 | 69 |
         }
853 |  |
854 |  |
         /**
855 |  |
          * Notify controller started
856 |  |
          */
857 |  |
         void notifyStarted() {
858 | 86 |
             for(ControllerStateListener stateListener : stateListeners) {
859 | 70 |
                 stateListener.onStarted();
860 |  |
             }
861 | 86 |
         }
862 |  |
863 |  |
864 |  |
         /**
865 |  |
          * Notify controller is ready
866 |  |
          */
867 |  |
         void notifyReady() {
868 | 136 |
             if (readySelectorHandlerCounter.decrementAndGet() == 0) {
869 | 87 |
                 for (ControllerStateListener stateListener : stateListeners) {
870 | 72 |
                     stateListener.onReady();
871 |  |
                 }
872 |  |
             }
873 | 136 |
         }
874 |  |
875 |  |
876 |  |
         /**
877 |  |
          * Notify controller stopped
878 |  |
          */
879 |  |
         void notifyStopped() {
880 | 136 |
             if (stoppedSelectorHandlerCounter.decrementAndGet() == 0) {
881 |  |
                 // Notify internal listeners
882 | 87 |
                 synchronized(stoppedSelectorHandlerCounter) {
883 | 87 |
                     stoppedSelectorHandlerCounter.notifyAll();
884 | 87 |
                 }
885 |  |
             }
886 | 136 |
         }
887 |  |
888 |  |
889 |  |
         /**
890 |  |
          * Notify exception occured
891 |  |
          */
892 |  |
         void notifyException(Throwable e) {
893 | 0 |
             for(ControllerStateListener stateListener : stateListeners) {
894 | 0 |
                 stateListener.onException(e);
895 |  |
             }
896 | 0 |
         }
897 |  |
898 |  |
899 |  |
         /**
900 |  |
          * Start the Controller. If the Pipeline and/or Handler has not been
901 |  |
          * defined, the default will be used.
902 |  |
          */
903 |  |
         public void start() throws IOException {
904 | 117 |
             stateHolder.getStateLocker().writeLock().lock();
905 | 117 |
             boolean isUnlocked = false;
906 |  |
             try {
907 | 117 |
                 if (stateHolder.getState(false) == null ||
908 |  |
                         stateHolder.getState(false) == State.STOPPED) {
909 |  |
                     // if selectorHandlers were not set by user explicitly,
910 |  |
                     // add TCPSelectorHandler by default
911 | 69 |
                     if (selectorHandlers.isEmpty()) {
912 | 2 |
                         SelectorHandler selectorHandler = new TCPSelectorHandler();
913 | 2 |
                         selectorHandlers.add(selectorHandler);
914 |  |
                     }
915 |  |
916 | 69 |
                     pipeline.initPipeline();
917 | 69 |
                     pipeline.startPipeline();
918 |  |
919 | 69 |
                     if (readThreadsCount > 0) {
920 | 6 |
                         initReadThreads();
921 | 6 |
                         multiReadThreadSelectorHandler =
922 |  |
                                 new RoundRobinSelectorHandler(readThreadControllers);
923 |  |
                     }
924 |  |
925 | 69 |
                     stateHolder.setState(State.STARTED, false);
926 | 69 |
                     notifyStarted();
927 |  |
928 | 69 |
                     int selectorHandlerCount = selectorHandlers.size();
929 | 69 |
                     readySelectorHandlerCounter = new AtomicInteger(selectorHandlerCount);
930 | 69 |
                     stoppedSelectorHandlerCounter = new AtomicInteger(selectorHandlerCount);
931 |  |
932 |  |
933 | 69 |
                     Iterator<SelectorHandler> it = selectorHandlers.iterator();
934 | 69 |
                     if (selectorHandlerCount > 1) {
935 | 4 |
                         for (; it.hasNext() && selectorHandlerCount-- > 0;) {
936 | 3 |
                             SelectorHandler selectorHandler = it.next();
937 | 3 |
                             startSelectorHandlerRunner(selectorHandler, true);
938 | 3 |
                         }
939 | 68 |
                     } else if (it.hasNext()) {
940 | 68 |
                         SelectorHandler selectorHandler = it.next();
941 | 68 |
                         stateHolder.getStateLocker().writeLock().unlock();
942 | 68 |
                         isUnlocked = true;
943 | 68 |
                         startSelectorHandlerRunner(selectorHandler, false);
944 |  |
                     }
945 |  |
                 }
946 |  |
             } finally {
947 | 117 |
                 if (!isUnlocked) {
948 | 49 |
                     stateHolder.getStateLocker().writeLock().unlock();
949 |  |
                 }
950 |  |
             }
951 |  |
952 | 117 |
             waitUntilSeletorHandlersStop();
953 |  |
954 | 117 |
             if (readThreadsCount > 0) {
955 | 6 |
                 multiReadThreadSelectorHandler.shutdown();
956 | 6 |
                 multiReadThreadSelectorHandler = null;
957 |  |
958 | 21 |
                 for (Controller readController : readThreadControllers) {
959 |  |
                     try {
960 | 15 |
                         readController.stop();
961 | 0 |
                     } catch (IOException e) {
962 | 0 |
                         logger.log(Level.WARNING, "Exception occured when stopping read Controller!", e);
963 | 15 |
                     }
964 |  |
                 }
965 |  |
966 | 6 |
                 readThreadControllers = null;
967 |  |
             }
968 |  |
969 | 117 |
             selectorHandlers.clear();
970 | 117 |
             pipeline.stopPipeline();
971 | 117 |
             attributes = null;
972 |  |
973 |  |
             // Notify Controller listeners
974 | 117 |
             for (ControllerStateListener stateListener : stateListeners) {
975 | 138 |
                 stateListener.onStopped();
976 |  |
             }
977 | 117 |
         }
978 |  |
979 |  |
980 |  |
         /**
981 |  |
          * Stop the Controller by canceling all the registered keys.
982 |  |
          */
983 |  |
         public void stop() throws IOException {
984 | 70 |
             stop(false);
985 | 70 |
         }
986 |  |
987 |  |
988 |  |
         /**
989 |  |
          * Stop the Controller by canceling all the registered keys.
990 |  |
          * @param isAsync, true if controller should be stopped asynchrounously and control
991 |  |
          *                 returned immediately. If false - control will be returned
992 |  |
          *                 after Controller will be completely stoped.
993 |  |
          */
994 |  |
         public void stop(boolean isAsync) throws IOException {
995 | 70 |
             final CountDownLatch latch = new CountDownLatch(1);
996 | 70 |
             stateHolder.getStateLocker().writeLock().lock();
997 |  |
             try {
998 | 70 |
                 if (stateHolder.getState(false) == State.STOPPED) {
999 | 1 |
                     logger.log(Level.FINE, "Controller is already in stopped state");
1000 |  |
                     return;
1001 |  |
                 }
1002 |  |
1003 | 69 |
                 if (!isAsync) {
1004 | 69 |
                     addStateListener(new ControllerStateListenerAdapter() {
1005 |  |
                         @Override
1006 |  |
                         public void onException(Throwable e) {
1007 | 0 |
                             removeStateListener(this);
1008 | 0 |
                             latch.countDown();
1009 | 0 |
                         }
1010 |  |
1011 |  |
                         @Override
1012 |  |
                         public void onStopped() {
1013 | 69 |
                             removeStateListener(this);
1014 | 69 |
                             latch.countDown();
1015 | 69 |
                         }
1016 |  |
1017 |  |
                     });
1018 |  |
                 }
1019 | 69 |
                 stateHolder.setState(State.STOPPED, false);
1020 |  |
             } finally {
1021 | 70 |
                 stateHolder.getStateLocker().writeLock().unlock();
1022 | 69 |
             }
1023 |  |
1024 | 69 |
             if (!isAsync) {
1025 |  |
                 try {
1026 | 69 |
                     latch.await();
1027 | 69 |
                 } catch(InterruptedException e) {}
1028 |  |
             }
1029 | 69 |
         }
1030 |  |
1031 |  |
1032 |  |
         /**
1033 |  |
          * Pause this {@link Controller} and associated {@link SelectorHandler}s
1034 |  |
          */
1035 |  |
         public void pause() throws IOException {
1036 | 0 |
             stateHolder.setState(State.PAUSED);
1037 | 0 |
         }
1038 |  |
1039 |  |
1040 |  |
         /**
1041 |  |
          * Resume this {@link Controller} and associated {@link SelectorHandler}s
1042 |  |
          */
1043 |  |
         public void resume() throws IOException {
1044 | 0 |
             if (!State.PAUSED.equals(stateHolder.getState(false))) {
1045 | 0 |
                 throw new IllegalStateException("Controller is not in PAUSED state, but: " +
1046 |  |
                         stateHolder.getState(false));
1047 |  |
             }
1048 |  |
1049 | 0 |
             stateHolder.setState(State.STARTED);
1050 | 0 |
         }
1051 |  |
1052 |  |
1053 |  |
         /**
1054 |  |
          * Gets this {@link Controller}'s {@link StateHolder}
1055 |  |
          * @return {@link StateHolder}
1056 |  |
          */
1057 |  |
         public StateHolder<State> getStateHolder() {
1058 | 136 |
             return stateHolder;
1059 |  |
         }
1060 |  |
1061 |  |
1062 |  |
         /**
1063 |  |
          * Initialize the number of ReadThreadController.
1064 |  |
          */
1065 |  |
         private void initReadThreads() throws IOException {
1066 |  |
             // Attributes need to be shared among Controller and its ReadControllers
1067 | 6 |
             if (attributes == null) {
1068 | 6 |
                 attributes = new HashMap<String, Object>(2);
1069 |  |
             }
1070 |  |
1071 | 6 |
             readThreadControllers = new ReadController[readThreadsCount];
1072 | 21 |
             for(int i=0; i<readThreadsCount; i++) {
1073 | 15 |
                 ReadController controller = new ReadController();
1074 | 15 |
                 copyTo(controller);
1075 | 15 |
                 controller.setReadThreadsCount(0);
1076 | 15 |
                 readThreadControllers[i] = controller;
1077 |  |
             }
1078 |  |
1079 | 6 |
             for (SelectorHandler selectorHandler : selectorHandlers) {
1080 | 6 |
                 addSelectorHandlerOnReadControllers(selectorHandler);
1081 |  |
             }
1082 |  |
1083 | 21 |
             for (int i=0; i < readThreadControllers.length; i++) {
1084 |  |
                 // TODO Get a Thread from a Pool instead.
1085 | 15 |
                 new Thread(readThreadControllers[i], "GrizzlyReadController-" + i).start();
1086 |  |
             }
1087 | 6 |
         }
1088 |  |
1089 |  |
1090 |  |
         /**
1091 |  |
          * Register {@link SelectorHandler} on all read controllers
1092 |  |
          * @param selectorHandler
1093 |  |
          */
1094 |  |
         private void addSelectorHandlerOnReadControllers(SelectorHandler selectorHandler) {
1095 | 54 |
             if (readThreadControllers == null || readThreadsCount == 0) return;
1096 |  |
1097 |  |
             // Attributes need to be shared among SelectorHandler and its read-copies
1098 | 6 |
             if (selectorHandler.getAttributes() == null) {
1099 | 6 |
                 selectorHandler.setAttributes(new HashMap<String, Object>(2));
1100 |  |
             }
1101 |  |
1102 | 21 |
             for (Controller readController : readThreadControllers) {
1103 | 15 |
                 SelectorHandler copySelectorHandler = Cloner.clone(selectorHandler);
1104 |  |
                 try {
1105 | 15 |
                     copySelectorHandler.setSelector(Selector.open());
1106 | 0 |
                 } catch(IOException e) {
1107 | 0 |
                     logger.log(Level.SEVERE, "Error opening selector!", e);
1108 | 15 |
                 }
1109 |  |
1110 | 15 |
                 readController.addSelectorHandler(copySelectorHandler);
1111 |  |
             }
1112 | 6 |
         }
1113 |  |
1114 |  |
1115 |  |
         /**
1116 |  |
          * Starts <code>SelectorHandlerRunner</code>
1117 |  |
          * @param selectorHandler
1118 |  |
          * @param isRunAsync if true - <code>SelectorHandlerRunner</code> will be run
1119 |  |
          *          in separate <code>Thread</code>, if false - in current <code>Thread</code>
1120 |  |
          */
1121 |  |
         private void startSelectorHandlerRunner(SelectorHandler selectorHandler,
1122 |  |
                 boolean isRunAsync) {
1123 |  |
1124 |  |
             // check if there is java.nio.Selector already open,
1125 |  |
             // if so, just notify the controller onReady() listeners
1126 | 119 |
             if (selectorHandler.getSelector() != null) {
1127 | 0 |
                 notifyReady();
1128 |  |
             }
1129 | 119 |
             Runnable selectorRunner = new SelectorHandlerRunner(this, selectorHandler);
1130 | 119 |
             if (isRunAsync) {
1131 |  |
                 // if there are more than 1 selector handler - run it in separate thread
1132 |  |
                 //@TODO Take Thread from ThreadPool?
1133 | 51 |
                 new Thread(selectorRunner, "GrizzlySelectorRunner-" + selectorHandler.protocol()).start();
1134 |  |
             } else {
1135 |  |
                 // else run it in current thread
1136 | 68 |
                 selectorRunner.run();
1137 |  |
             }
1138 | 119 |
         }
1139 |  |
1140 |  |
1141 |  |
         /**
1142 |  |
          * Register {@link SelectorHandler} on all read controllers
1143 |  |
          * @param selectorHandler
1144 |  |
          */
1145 |  |
         private void removeSelectorHandlerOnReadControllers(SelectorHandler selectorHandler) {
1146 | 1 |
             if (readThreadControllers == null) return;
1147 |  |
1148 | 0 |
             for (ReadController readController : readThreadControllers) {
1149 | 0 |
                 readController.removeSelectorHandlerClone(selectorHandler);
1150 |  |
             }
1151 | 0 |
         }
1152 |  |
1153 |  |
1154 |  |
         /**
1155 |  |
          * Is this Controller started?
1156 |  |
          * @return <code>boolean</code> true / false
1157 |  |
          */
1158 |  |
         public boolean isStarted() {
1159 | 39 |
             return stateHolder.getState() == State.STARTED;
1160 |  |
         }
1161 |  |
1162 |  |
1163 |  |
         // ----------- ConnectorHandlerPool interface implementation ----------- //
1164 |  |
1165 |  |
1166 |  |
         /**
1167 |  |
          * Return an instance of a {@link ConnectorHandler} based on the
1168 |  |
          * Protocol requested.
1169 |  |
          */
1170 |  |
         public ConnectorHandler acquireConnectorHandler(Protocol protocol){
1171 | 184 |
             return connectorHandlerPool.acquireConnectorHandler(protocol);
1172 |  |
         }
1173 |  |
1174 |  |
1175 |  |
         /**
1176 |  |
          * Return a {@link ConnectorHandler} to the pool of ConnectorHandler.
1177 |  |
          * Any reference to the returned must not be re-used as that instance
1178 |  |
          * can always be acquired again, causing unexpected results.
1179 |  |
          */
1180 |  |
         public void releaseConnectorHandler(ConnectorHandler connectorHandler){
1181 | 184 |
             connectorHandlerPool.releaseConnectorHandler(connectorHandler);
1182 | 184 |
         }
1183 |  |
1184 |  |
1185 |  |
         /**
1186 |  |
          * <tt>true</tt> if OP_ERAD and OP_WRITE can be handled concurrently.
1187 |  |
          * If <tt>false</tt>, the Controller will first invoke the OP_READ handler and
1188 |  |
          * then invoke the OP_WRITE during the next Selector.select() invocation.
1189 |  |
          */
1190 |  |
         public boolean isHandleReadWriteConcurrently() {
1191 | 0 |
             return handleReadWriteConcurrently;
1192 |  |
         }
1193 |  |
1194 |  |
1195 |  |
         /**
1196 |  |
          * <tt>true</tt> if OP_ERAD and OP_WRITE can be handled concurrently.
1197 |  |
          * If <tt>false</tt>, the Controller will first invoke the OP_READ handler and
1198 |  |
          * then invoke the OP_WRITE during the next Selector.select() invocation.
1199 |  |
          */
1200 |  |
         public void setHandleReadWriteConcurrently(boolean handleReadWriteConcurrently) {
1201 | 7 |
             this.handleReadWriteConcurrently = handleReadWriteConcurrently;
1202 | 7 |
         }
1203 |  |
1204 |  |
         /**
1205 |  |
          * Method waits until all initialized {@link SelectorHandler}s will
1206 |  |
          * not get stopped
1207 |  |
          */
1208 |  |
         protected void waitUntilSeletorHandlersStop() {
1209 | 151 |
             synchronized(stoppedSelectorHandlerCounter) {
1210 | 742 |
                 while(stoppedSelectorHandlerCounter.get() > 0 ||
1211 |  |
                         !State.STOPPED.equals(stateHolder.getState())) {
1212 |  |
                     try {
1213 | 591 |
                         stoppedSelectorHandlerCounter.wait(1000);
1214 | 0 |
                     } catch (InterruptedException ex) {
1215 | 591 |
                     }
1216 |  |
                 }
1217 | 151 |
             }
1218 | 151 |
         }
1219 |  |
1220 |  |
         // ----------- AttributeHolder interface implementation ----------- //
1221 |  |
1222 |  |
         /**
1223 |  |
          * Remove a key/value object.
1224 |  |
          * Method is not thread safe
1225 |  |
          *
1226 |  |
          * @param key - name of an attribute
1227 |  |
          * @return  attribute which has been removed
1228 |  |
          */
1229 |  |
         public Object removeAttribute(String key) {
1230 | 0 |
             if (attributes == null) return null;
1231 |  |
1232 | 0 |
             return attributes.remove(key);
1233 |  |
         }
1234 |  |
1235 |  |
         /**
1236 |  |
          * Set a key/value object.
1237 |  |
          * Method is not thread safe
1238 |  |
          *
1239 |  |
          * @param key - name of an attribute
1240 |  |
          * @param value - value of named attribute
1241 |  |
          */
1242 |  |
         public void setAttribute(String key, Object value) {
1243 | 0 |
             if (attributes == null) {
1244 | 0 |
                 attributes = new HashMap<String, Object>();
1245 |  |
             }
1246 |  |
1247 | 0 |
             attributes.put(key, value);
1248 | 0 |
         }
1249 |  |
1250 |  |
         /**
1251 |  |
          * Return an object based on a key.
1252 |  |
          * Method is not thread safe
1253 |  |
          *
1254 |  |
          * @param key - name of an attribute
1255 |  |
          * @return - attribute value for the <tt>key</tt>, null if <tt>key</tt>
1256 |  |
          *           does not exist in <tt>attributes</tt>
1257 |  |
          */
1258 |  |
         public Object getAttribute(String key) {
1259 | 0 |
             if (attributes == null) return null;
1260 |  |
1261 | 0 |
             return attributes.get(key);
1262 |  |
         }
1263 |  |
1264 |  |
         /**
1265 |  |
          * Set a {@link Map} of attribute name/value pairs.
1266 |  |
          * Old {@link AttributeHolder} values will not be available.
1267 |  |
          * Later changes of this {@link Map} will lead to changes to the current
1268 |  |
          * {@link AttributeHolder}.
1269 |  |
          *
1270 |  |
          * @param attributes - map of name/value pairs
1271 |  |
          */
1272 |  |
         public void setAttributes(Map<String, Object> attributes) {
1273 | 0 |
             this.attributes = attributes;
1274 | 0 |
         }
1275 |  |
1276 |  |
1277 |  |
         /**
1278 |  |
          * Return a {@link Map} of attribute name/value pairs.
1279 |  |
          * Updates, performed on the returned {@link Map} will be reflected in
1280 |  |
          * this {@link AttributeHolder}
1281 |  |
          *
1282 |  |
          * @return - {@link Map} of attribute name/value pairs
1283 |  |
          */
1284 |  |
         public Map<String, Object> getAttributes() {
1285 | 0 |
             return attributes;
1286 |  |
         }
1287 |  |
1288 |  |
1289 |  |
         /**
1290 |  |
          * Return the Controller which is handling the {@link Handler}
1291 |  |
          * @param handler The handler (like {@link SelectorHandler})
1292 |  |
          * @return The Controller associated with the Handler, or null if not
1293 |  |
          * associated.
1294 |  |
          */
1295 |  |
         public static Controller getHandlerController(Handler handler){
1296 | 175 |
             if (handler instanceof SelectorHandler){
1297 | 175 |
                 for (Controller controller: controllers){
1298 | 7486 |
                     if (controller.getSelectorHandlers().contains(handler)){
1299 | 175 |
                         return controller;
1300 |  |
                     }
1301 |  |
                 }
1302 |  |
             }
1303 | 0 |
             return null;
1304 |  |
         }
1305 |  |
     }