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

Classes in this File | Line Coverage | Branch Coverage | Complexity
TCPSelectorHandler |
74 % | 287/387
65 % | 81/124
| 0.0;0
TCPSelectorHandler$1 |
100 % | 2/2
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.async.AsyncQueueReader;
42 |  |
     import com.sun.grizzly.async.AsyncQueueReaderContextTask;
43 |  |
     import com.sun.grizzly.async.AsyncQueueWriter;
44 |  |
     import com.sun.grizzly.async.TCPAsyncQueueWriter;
45 |  |
     import com.sun.grizzly.async.AsyncQueueWriterContextTask;
46 |  |
     import com.sun.grizzly.async.TCPAsyncQueueReader;
47 |  |
     import com.sun.grizzly.util.CallbackHandlerSelectionKeyAttachment;
48 |  |
     import com.sun.grizzly.util.Cloner;
49 |  |
     import com.sun.grizzly.util.Copyable;
50 |  |
     import com.sun.grizzly.util.SelectionKeyAttachment;
51 |  |
     import com.sun.grizzly.util.SelectionKeyOP;
52 |  |
     import com.sun.grizzly.util.SelectionKeyOP.ConnectSelectionKeyOP;
53 |  |
     import com.sun.grizzly.util.State;
54 |  |
     import com.sun.grizzly.util.StateHolder;
55 |  |
     import java.io.IOException;
56 |  |
     import java.net.BindException;
57 |  |
     import java.net.InetAddress;
58 |  |
     import java.net.ServerSocket;
59 |  |
     import java.net.Socket;
60 |  |
     import java.net.InetSocketAddress;
61 |  |
     import java.net.SocketAddress;
62 |  |
     import java.net.SocketException;
63 |  |
     import java.nio.channels.CancelledKeyException;
64 |  |
     import java.nio.channels.ClosedChannelException;
65 |  |
     import java.nio.channels.ClosedSelectorException;
66 |  |
     import java.nio.channels.SelectableChannel;
67 |  |
     import java.nio.channels.SelectionKey;
68 |  |
     import java.nio.channels.Selector;
69 |  |
     import java.nio.channels.ServerSocketChannel;
70 |  |
     import java.nio.channels.SocketChannel;
71 |  |
     import java.util.ConcurrentModificationException;
72 |  |
     import java.util.HashMap;
73 |  |
     import java.util.Iterator;
74 |  |
     import java.util.Map;
75 |  |
     import java.util.Set;
76 |  |
     import java.util.concurrent.Callable;
77 |  |
     import java.util.concurrent.ConcurrentLinkedQueue;
78 |  |
     import java.util.concurrent.atomic.AtomicBoolean;
79 |  |
     import java.util.logging.Level;
80 |  |
     import java.util.logging.Logger;
81 |  |
82 |  |
     /**
83 |  |
      * A SelectorHandler handles all java.nio.channels.Selector operations.
84 |  |
      * One or more instance of a Selector are handled by SelectorHandler.
85 |  |
      * The logic for processing of SelectionKey interest (OP_ACCEPT,OP_READ, etc.)
86 |  |
      * is usually defined using an instance of SelectorHandler.
87 |  |
      *
88 |  |
      * This class represents a TCP implementation of a SelectorHandler.
89 |  |
      * This class first bind a ServerSocketChannel to a TCP port and then start
90 |  |
      * waiting for NIO events.
91 |  |
      *
92 |  |
      * @author Jeanfrancois Arcand
93 |  |
      */
94 |  |
     public class TCPSelectorHandler implements SelectorHandler {
95 |  |
96 |  |
97 |  |
         /**
98 |  |
          * The ConnectorInstanceHandler used to return a new or pooled
99 |  |
          * ConnectorHandler
100 |  |
          */
101 |  |
         protected ConnectorInstanceHandler connectorInstanceHandler;
102 |  |
103 |  |
104 |  |
         /**
105 |  |
          * The list of {@link SelectionKeyOP} to register next time the
106 |  |
          * Selector.select is invoked.
107 |  |
          */
108 |  |
         protected ConcurrentLinkedQueue<SelectionKeyOP> opToRegister;
109 |  |
110 |  |
         /**
111 |  |
          * The socket tcpDelay.
112 |  |
          *
113 |  |
          * Default value for tcpNoDelay is disabled (set to true).
114 |  |
          */
115 | 141 |
         protected boolean tcpNoDelay = true;
116 |  |
117 |  |
118 |  |
         /**
119 |  |
          * The socket reuseAddress
120 |  |
          */
121 | 141 |
         protected boolean reuseAddress = true;
122 |  |
123 |  |
124 |  |
         /**
125 |  |
          * The socket linger.
126 |  |
          */
127 | 141 |
         protected int linger = -1;
128 |  |
129 |  |
130 |  |
         /**
131 |  |
          * The socket time out
132 |  |
          */
133 | 141 |
         protected int socketTimeout = -1;
134 |  |
135 |  |
136 |  |
         protected Logger logger;
137 |  |
138 |  |
139 |  |
         /**
140 |  |
          * The server socket time out
141 |  |
          */
142 | 141 |
         protected int serverTimeout = 0;
143 |  |
144 |  |
145 |  |
         /**
146 |  |
          * The inet address to use when binding.
147 |  |
          */
148 |  |
         protected InetAddress inet;
149 |  |
150 |  |
151 |  |
         /**
152 |  |
          * The default TCP port.
153 |  |
          */
154 | 141 |
         protected int port = 18888;
155 |  |
156 |  |
157 |  |
         /**
158 |  |
          * The ServerSocket instance.
159 |  |
          */
160 |  |
         protected ServerSocket serverSocket;
161 |  |
162 |  |
163 |  |
         /**
164 |  |
          * The ServerSocketChannel.
165 |  |
          */
166 |  |
         protected ServerSocketChannel serverSocketChannel;
167 |  |
168 |  |
169 |  |
         /**
170 |  |
          * The single Selector.
171 |  |
          */
172 |  |
         protected Selector selector;
173 |  |
174 |  |
175 |  |
         /**
176 |  |
          * The Selector time out.
177 |  |
          */
178 | 141 |
         protected long selectTimeout = 1000L;
179 |  |
180 |  |
181 |  |
         /**
182 |  |
          * Server socket backlog.
183 |  |
          */
184 | 141 |
         protected int ssBackLog = 4096;
185 |  |
186 |  |
187 |  |
         /**
188 |  |
          * Is this used for client only or client/server operation.
189 |  |
          */
190 | 141 |
         protected Role role = Role.CLIENT_SERVER;
191 |  |
192 |  |
193 |  |
         /**
194 |  |
          * The SelectionKeyHandler associated with this SelectorHandler.
195 |  |
          */
196 |  |
         protected SelectionKeyHandler selectionKeyHandler;
197 |  |
198 |  |
199 |  |
         /**
200 |  |
          * The ProtocolChainInstanceHandler used by this instance. If not set, and instance
201 |  |
          * of the DefaultInstanceHandler will be created.
202 |  |
          */
203 |  |
         protected ProtocolChainInstanceHandler instanceHandler;
204 |  |
205 |  |
206 |  |
         /**
207 |  |
          * The {@link Pipeline} used by this instance. If null -
208 |  |
          * {@link Controller}'s {@link Pipeline} will be used
209 |  |
          */
210 |  |
         protected Pipeline pipeline;
211 |  |
212 |  |
213 |  |
         /**
214 |  |
          * {@link AsyncQueueWriter}
215 |  |
          */
216 |  |
         protected AsyncQueueWriter asyncQueueWriter;
217 |  |
218 |  |
219 |  |
         /**
220 |  |
          * {@link AsyncQueueWriter}
221 |  |
          */
222 |  |
         protected AsyncQueueReader asyncQueueReader;
223 |  |
224 |  |
225 |  |
         /**
226 |  |
          * Attributes, associated with the {@link SelectorHandler} instance
227 |  |
          */
228 |  |
         protected Map<String, Object> attributes;
229 |  |
230 |  |
         /**
231 |  |
          * This {@link SelectorHandler} StateHolder, which is shared among
232 |  |
          * SelectorHandler and its clones
233 |  |
          */
234 | 141 |
         protected StateHolder<State> stateHolder = new StateHolder<State>(true);
235 |  |
236 |  |
         /**
237 |  |
          * Flag, which shows whether shutdown was called for this {@link SelectorHandler}
238 |  |
          */
239 | 141 |
         protected AtomicBoolean isShutDown = new AtomicBoolean(false);
240 |  |
241 |  |
         public TCPSelectorHandler(){
242 | 56 |
             this(Role.CLIENT_SERVER);
243 | 56 |
         }
244 |  |
245 |  |
246 |  |
         /**
247 |  |
          * Create a TCPSelectorHandler only used with ConnectorHandler.
248 |  |
          *
249 |  |
          * @param isClient true if this SelectorHandler is only used
250 |  |
          * to handle ConnectorHandler.
251 |  |
          */
252 |  |
         public TCPSelectorHandler(boolean isClient) {
253 | 79 |
             this(boolean2Role(isClient));
254 | 79 |
         }
255 |  |
256 |  |
257 |  |
         /**
258 |  |
          * Create a TCPSelectorHandler only used with ConnectorHandler.
259 |  |
          *
260 |  |
          * @param role the <tt>TCPSelectorHandler</tt> {@link Role}
261 |  |
          */
262 | 141 |
         public TCPSelectorHandler(Role role) {
263 | 141 |
             this.role = role;
264 | 141 |
             logger = Controller.logger();
265 | 141 |
         }
266 |  |
267 |  |
         public void copyTo(Copyable copy) {
268 | 15 |
             TCPSelectorHandler copyHandler = (TCPSelectorHandler) copy;
269 | 15 |
             copyHandler.selector = selector;
270 | 15 |
             if (selectionKeyHandler != null) {
271 | 4 |
                 copyHandler.setSelectionKeyHandler(Cloner.clone(selectionKeyHandler));
272 |  |
             }
273 |  |
274 | 15 |
             copyHandler.attributes = attributes;
275 | 15 |
             copyHandler.selectTimeout = selectTimeout;
276 | 15 |
             copyHandler.serverTimeout = serverTimeout;
277 | 15 |
             copyHandler.inet = inet;
278 | 15 |
             copyHandler.port = port;
279 | 15 |
             copyHandler.ssBackLog = ssBackLog;
280 | 15 |
             copyHandler.tcpNoDelay = tcpNoDelay;
281 | 15 |
             copyHandler.linger = linger;
282 | 15 |
             copyHandler.socketTimeout = socketTimeout;
283 | 15 |
             copyHandler.logger = logger;
284 | 15 |
             copyHandler.reuseAddress = reuseAddress;
285 | 15 |
             copyHandler.connectorInstanceHandler = connectorInstanceHandler;
286 | 15 |
             copyHandler.stateHolder = stateHolder;
287 | 15 |
         }
288 |  |
289 |  |
290 |  |
         /**
291 |  |
          * Return the set of SelectionKey registered on this Selector.
292 |  |
          */
293 |  |
         public Set<SelectionKey> keys(){
294 | 266289 |
             if (selector != null){
295 | 266289 |
                 return selector.keys();
296 |  |
             } else {
297 | 0 |
                 throw new IllegalStateException("Selector is not created!");
298 |  |
             }
299 |  |
         }
300 |  |
301 |  |
302 |  |
         /**
303 |  |
          * Is the Selector open.
304 |  |
          */
305 |  |
         public boolean isOpen(){
306 | 265630 |
             if (selector != null){
307 | 265630 |
                 return selector.isOpen();
308 |  |
             } else {
309 | 0 |
                 return false;
310 |  |
             }
311 |  |
         }
312 |  |
313 |  |
314 |  |
         /**
315 |  |
          * Before invoking {@link Selector#select}, make sure the {@link ServerSocketChannel}
316 |  |
          * has been created. If true, then register all {@link SelectionKey} to the {@link Selector}.
317 |  |
          * @param ctx {@link Context}
318 |  |
          */
319 |  |
         public void preSelect(Context ctx) throws IOException {
320 | 215101 |
             initOpRegistriesIfRequired();
321 |  |
322 | 215101 |
             if (asyncQueueReader == null) {
323 | 130 |
                 asyncQueueReader = new TCPAsyncQueueReader(this);
324 |  |
             }
325 |  |
326 | 215101 |
             if (asyncQueueWriter == null) {
327 | 130 |
                 asyncQueueWriter = new TCPAsyncQueueWriter(this);
328 |  |
             }
329 |  |
330 | 215101 |
             if (selector == null){
331 |  |
                 try {
332 | 113 |
                     isShutDown.set(false);
333 |  |
334 | 113 |
                     connectorInstanceHandler = new ConnectorInstanceHandler.
335 |  |
                             ConcurrentQueueDelegateCIH(
336 |  |
                             getConnectorInstanceHandlerDelegate());
337 |  |
338 |  |
                     // Create the socket listener
339 | 113 |
                     selector = Selector.open();
340 |  |
341 | 113 |
                     if (role != Role.CLIENT){
342 | 34 |
                         serverSocketChannel = ServerSocketChannel.open();
343 | 34 |
                         serverSocket = serverSocketChannel.socket();
344 | 34 |
                         serverSocket.setReuseAddress(reuseAddress);
345 | 34 |
                         if ( inet == null){
346 | 34 |
                             serverSocket.bind(new InetSocketAddress(port),ssBackLog);
347 |  |
                         } else {
348 | 0 |
                             serverSocket.bind(new InetSocketAddress(inet,port),ssBackLog);
349 |  |
                         }
350 |  |
351 | 34 |
                         serverSocketChannel.configureBlocking(false);
352 | 34 |
                         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
353 |  |
                     }
354 | 113 |
                     ctx.getController().notifyReady();
355 | 0 |
                 } catch (SocketException ex){
356 | 0 |
                     throw new BindException(ex.getMessage() + ": " + port + "=" + this);
357 | 113 |
                 }
358 |  |
359 | 113 |
                 if (role != Role.CLIENT){
360 | 34 |
                     serverSocket.setSoTimeout(serverTimeout);
361 |  |
                 }
362 |  |
             } else {
363 | 214988 |
                 processPendingOperations(ctx);
364 |  |
             }
365 | 215101 |
         }
366 |  |
367 |  |
368 |  |
         protected void processPendingOperations(Context ctx) throws IOException {
369 | 266170 |
             if (!opToRegister.isEmpty()) {
370 |  |
                 SelectionKeyOP operation;
371 | 137761 |
                 Iterator<SelectionKeyOP> opIterator = opToRegister.iterator();
372 | 677725 |
                 while(opIterator.hasNext()) {
373 | 539964 |
                     operation = opIterator.next();
374 | 539964 |
                     opIterator.remove();
375 |  |
376 | 539964 |
                     if ((operation.getOp() & SelectionKey.OP_CONNECT) != 0) {
377 | 205 |
                         onConnectOp(ctx,
378 |  |
                                 (SelectionKeyOP.ConnectSelectionKeyOP) operation);
379 |  |
                     } else {
380 | 539759 |
                         if ((operation.getOp() & SelectionKey.OP_READ) != 0) {
381 | 158833 |
                             onReadOp(operation);
382 |  |
                         }
383 |  |
384 | 539759 |
                         if ((operation.getOp() & SelectionKey.OP_WRITE) != 0) {
385 | 380926 |
                             onWriteOp(operation);
386 |  |
                         }
387 |  |
                     }
388 |  |
389 | 539964 |
                     SelectionKeyOP.releaseSelectionKeyOP(operation);
390 |  |
                 }
391 |  |
             }
392 | 266170 |
         }
393 |  |
394 |  |
         /**
395 |  |
          * Handle new OP_READ ops.
396 |  |
          */
397 |  |
         protected void onReadOp(SelectionKeyOP selectionKeyOp)
398 |  |
                 throws ClosedChannelException {
399 | 158833 |
             SelectionKey key = selectionKeyOp.getSelectionKey();
400 | 158833 |
             if (key != null) {
401 | 158810 |
                 selectionKeyHandler.register(key, SelectionKey.OP_READ);
402 |  |
             } else {
403 | 23 |
                 selectionKeyHandler.register(selectionKeyOp.getChannel(),
404 |  |
                         SelectionKey.OP_READ);
405 |  |
             }
406 | 158833 |
         }
407 |  |
408 |  |
409 |  |
         /**
410 |  |
          * Handle new OP_WRITE ops.
411 |  |
          */
412 |  |
         protected void onWriteOp(SelectionKeyOP selectionKeyOp)
413 |  |
                 throws ClosedChannelException {
414 | 380926 |
             SelectionKey key = selectionKeyOp.getSelectionKey();
415 | 380926 |
             if (key != null) {
416 | 380926 |
                 selectionKeyHandler.register(key, SelectionKey.OP_WRITE);
417 |  |
             } else {
418 | 0 |
                 selectionKeyHandler.register(selectionKeyOp.getChannel(),
419 |  |
                         SelectionKey.OP_WRITE);
420 |  |
             }
421 | 380926 |
         }
422 |  |
423 |  |
424 |  |
         /**
425 |  |
          * Handle new OP_CONNECT ops.
426 |  |
          */
427 |  |
         protected void onConnectOp(Context ctx,
428 |  |
                 SelectionKeyOP.ConnectSelectionKeyOP selectionKeyOp) throws IOException {
429 | 174 |
             SocketAddress remoteAddress = selectionKeyOp.getRemoteAddress();
430 | 174 |
             SocketAddress localAddress = selectionKeyOp.getLocalAddress();
431 | 174 |
             CallbackHandler callbackHandler = selectionKeyOp.getCallbackHandler();
432 |  |
433 | 174 |
             SocketChannel socketChannel = SocketChannel.open();
434 | 174 |
             socketChannel.socket().setReuseAddress(reuseAddress);
435 | 174 |
             if (localAddress != null) {
436 | 0 |
                 socketChannel.socket().bind(localAddress);
437 |  |
             }
438 |  |
439 | 174 |
             socketChannel.configureBlocking(false);
440 | 174 |
             SelectionKey key = socketChannel.register(selector,
441 |  |
                     SelectionKey.OP_CONNECT);
442 | 174 |
             key.attach(CallbackHandlerSelectionKeyAttachment.create(key, callbackHandler));
443 |  |
444 |  |
             boolean isConnected;
445 |  |
             try {
446 | 174 |
                 isConnected = socketChannel.connect(remoteAddress);
447 | 0 |
             } catch(Exception e) {
448 | 0 |
                 if (logger.isLoggable(Level.FINE)) {
449 | 0 |
                     logger.log(Level.FINE, "Exception occured when tried to connect socket", e);
450 |  |
                 }
451 |  |
452 |  |
                 // set isConnected to true to let callback handler to know about the problem happened
453 | 0 |
                 isConnected = true;
454 | 174 |
             }
455 |  |
456 |  |
             // if channel was connected immediately or exception occured
457 | 174 |
             if (isConnected) {
458 | 0 |
                 onConnectInterest(key, ctx);
459 |  |
             }
460 | 174 |
         }
461 |  |
462 |  |
463 |  |
         /**
464 |  |
          * Execute the Selector.select(...) operations.
465 |  |
          * @param ctx {@link Context}
466 |  |
          * @return {@link Set} of {@link SelectionKey}
467 |  |
          */
468 |  |
         public Set<SelectionKey> select(Context ctx) throws IOException{
469 | 266289 |
             selector.select(selectTimeout);
470 | 266289 |
             return selector.selectedKeys();
471 |  |
         }
472 |  |
473 |  |
474 |  |
         /**
475 |  |
          * Invoked after Selector.select().
476 |  |
          * @param ctx {@link Context}
477 |  |
          */
478 |  |
         public void postSelect(Context ctx) {
479 | 266289 |
             Set<SelectionKey> readyKeys = keys();
480 | 266289 |
             if (readyKeys.isEmpty()){
481 | 659 |
                 return;
482 |  |
             }
483 |  |
484 | 265629 |
             if (isOpen()) {
485 | 265630 |
                 selectionKeyHandler.expire(readyKeys.iterator());
486 |  |
             }
487 | 265630 |
         }
488 |  |
489 |  |
490 |  |
         /**
491 |  |
          * Register a SelectionKey to this Selector.
492 |  |
          */
493 |  |
         public void register(SelectionKey key, int ops) {
494 | 539750 |
             if (key == null) {
495 | 0 |
                 throw new NullPointerException("SelectionKey parameter is null");
496 |  |
             }
497 |  |
498 | 539750 |
             SelectionKeyOP keyOP = SelectionKeyOP.aquireSelectionKeyOP(ops);
499 | 539750 |
             keyOP.setSelectionKey(key);
500 | 539748 |
             keyOP.setOp(ops);
501 | 539749 |
             initOpRegistriesIfRequired();
502 | 539749 |
             opToRegister.offer(keyOP);
503 | 539749 |
             selector.wakeup();
504 | 539749 |
         }
505 |  |
506 |  |
         public void register(SelectableChannel channel, int ops) {
507 | 23 |
             if (channel == null) {
508 | 0 |
                 throw new NullPointerException("SelectableChannel parameter is null");
509 |  |
             }
510 |  |
511 | 23 |
             SelectionKeyOP keyOP = SelectionKeyOP.aquireSelectionKeyOP(ops);
512 | 23 |
             keyOP.setChannel(channel);
513 | 23 |
             keyOP.setOp(ops);
514 | 23 |
             initOpRegistriesIfRequired();
515 | 23 |
             opToRegister.offer(keyOP);
516 | 23 |
             selector.wakeup();
517 | 23 |
         }
518 |  |
519 |  |
         /**
520 |  |
          * Register a CallBackHandler to this Selector.
521 |  |
          *
522 |  |
          * @param remoteAddress remote address to connect
523 |  |
          * @param localAddress local address to bin
524 |  |
          * @param callBackHandler {@link CallbackHandler}
525 |  |
          * @throws java.io.IOException
526 |  |
          */
527 |  |
         protected void connect(SocketAddress remoteAddress, SocketAddress localAddress,
528 |  |
                 CallbackHandler callBackHandler) throws IOException {
529 |  |
530 | 205 |
             SelectionKeyOP.ConnectSelectionKeyOP keyOP =
531 |  |
                     (ConnectSelectionKeyOP) SelectionKeyOP.aquireSelectionKeyOP(SelectionKey.OP_CONNECT);
532 |  |
533 | 205 |
             keyOP.setOp(SelectionKey.OP_CONNECT);
534 | 205 |
             keyOP.setRemoteAddress(remoteAddress);
535 | 205 |
             keyOP.setLocalAddress(localAddress);
536 | 205 |
             keyOP.setCallbackHandler(callBackHandler);
537 | 205 |
             initOpRegistriesIfRequired();
538 | 205 |
             opToRegister.offer(keyOP);
539 | 205 |
             selector.wakeup();
540 | 205 |
         }
541 |  |
542 |  |
         /**
543 |  |
          * {@inheritDoc}
544 |  |
          */
545 |  |
         public void pause() {
546 | 1 |
             stateHolder.setState(State.PAUSED);
547 | 1 |
         }
548 |  |
549 |  |
         /**
550 |  |
          * {@inheritDoc}
551 |  |
          */
552 |  |
         public void resume() {
553 | 1 |
             if (!State.PAUSED.equals(stateHolder.getState(false))) {
554 | 0 |
                 throw new IllegalStateException("SelectorHandler is not in PAUSED state, but: " +
555 |  |
                         stateHolder.getState(false));
556 |  |
             }
557 |  |
558 | 1 |
             stateHolder.setState(State.STARTED);
559 | 1 |
         }
560 |  |
561 |  |
         /**
562 |  |
          * {@inheritDoc}
563 |  |
          */
564 |  |
         public StateHolder<State> getStateHolder() {
565 | 266473 |
             return stateHolder;
566 |  |
         }
567 |  |
568 |  |
         /**
569 |  |
          * Shuntdown this instance by closing its Selector and associated channels.
570 |  |
          */
571 |  |
         public void shutdown() {
572 |  |
             // If shutdown was called for this SelectorHandler
573 | 137 |
             if (isShutDown.getAndSet(true)) return;
574 |  |
575 | 136 |
             stateHolder.setState(State.STOPPED);
576 |  |
577 | 136 |
             if (selector != null) {
578 |  |
                 try {
579 | 130 |
                     for (SelectionKey selectionKey : selector.keys()) {
580 | 76 |
                         selectionKeyHandler.close(selectionKey);
581 |  |
                     }
582 | 0 |
                 } catch (ClosedSelectorException e) {
583 |  |
                     // If Selector is already closed - OK
584 | 0 |
                 } catch (ConcurrentModificationException e) {
585 |  |
                     // Someone else works with keys. Create copy
586 | 0 |
                     Object[] keys = selector.keys().toArray();
587 | 0 |
                     for (Object selectionKey : keys) {
588 | 0 |
                         selectionKeyHandler.close((SelectionKey) selectionKey);
589 |  |
                     }
590 |  |
591 | 130 |
                 }
592 |  |
             }
593 |  |
594 |  |
             try{
595 | 136 |
                 if (serverSocket != null)
596 | 34 |
                     serverSocket.close();
597 | 0 |
             } catch (Throwable ex){
598 | 0 |
                 Controller.logger().log(Level.SEVERE,
599 |  |
                         "serverSocket.close",ex);
600 | 136 |
             }
601 |  |
602 |  |
             try{
603 | 136 |
                 if (serverSocketChannel != null)
604 | 34 |
                     serverSocketChannel.close();
605 | 0 |
             } catch (Throwable ex){
606 | 0 |
                 Controller.logger().log(Level.SEVERE,
607 |  |
                         "serverSocketChannel.close",ex);
608 | 136 |
             }
609 |  |
610 |  |
             try{
611 | 136 |
                 if (selector != null)
612 | 130 |
                     selector.close();
613 | 0 |
             } catch (Throwable ex){
614 | 0 |
                 Controller.logger().log(Level.SEVERE,
615 |  |
                         "selector.close",ex);
616 | 136 |
             }
617 |  |
618 | 136 |
             if (asyncQueueReader != null) {
619 | 130 |
                 asyncQueueReader.close();
620 | 130 |
                 asyncQueueReader = null;
621 |  |
             }
622 |  |
623 | 136 |
             if (asyncQueueWriter != null) {
624 | 130 |
                 asyncQueueWriter.close();
625 | 130 |
                 asyncQueueWriter = null;
626 |  |
             }
627 |  |
628 | 136 |
             attributes = null;
629 | 136 |
         }
630 |  |
631 |  |
         /**
632 |  |
          * {@inheritDoc}
633 |  |
          */
634 |  |
         public SelectableChannel acceptWithoutRegistration(SelectionKey key)
635 |  |
         throws IOException {
636 | 199 |
             ServerSocketChannel server = (ServerSocketChannel) key.channel();
637 | 199 |
             SocketChannel channel = server.accept();
638 | 199 |
             return channel;
639 |  |
         }
640 |  |
641 |  |
         /**
642 |  |
          * Handle OP_ACCEPT.
643 |  |
          * @param ctx {@link Context}
644 |  |
          * @return always returns false
645 |  |
          */
646 |  |
         public boolean onAcceptInterest(SelectionKey key,
647 |  |
                 Context ctx) throws IOException{
648 | 178 |
             SelectableChannel channel = acceptWithoutRegistration(key);
649 |  |
650 | 178 |
             if (channel != null) {
651 | 178 |
                 configureChannel(channel);
652 | 178 |
                 SelectionKey readKey =
653 |  |
                         channel.register(selector, SelectionKey.OP_READ);
654 | 178 |
                 readKey.attach(System.currentTimeMillis());
655 |  |
             }
656 | 178 |
             return false;
657 |  |
         }
658 |  |
659 |  |
         /**
660 |  |
          * Handle OP_READ.
661 |  |
          * @param ctx {@link Context}
662 |  |
          * @param key {@link SelectionKey}
663 |  |
          * @return false if handled by a {@link CallbackHandler}, otherwise true
664 |  |
          */
665 |  |
         public boolean onReadInterest(final SelectionKey key,final Context ctx)
666 |  |
             throws IOException{
667 |  |
             // disable OP_READ on key before doing anything else
668 | 158935 |
             key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
669 | 158935 |
             Object attach = SelectionKeyAttachment.getAttachment(key);
670 |  |
671 | 158935 |
             if (asyncQueueReader.isAsyncQueueReaderEnabledFor(key)) {
672 | 9757 |
                 final Context context = pollContext(ctx, key, Context.OpType.OP_READ);
673 | 9757 |
                 invokeAsyncQueueReader(context);
674 | 9757 |
                 return false;
675 | 149178 |
             } else if (attach instanceof CallbackHandler){
676 | 58379 |
                 final Context context = pollContext(ctx, key, Context.OpType.OP_READ);
677 | 58379 |
                 invokeCallbackHandler((CallbackHandler) attach, context);
678 | 58379 |
                 return false;
679 |  |
             } else {
680 | 90799 |
                 return true;
681 |  |
             }
682 |  |
         }
683 |  |
684 |  |
685 |  |
         /**
686 |  |
          * Handle OP_WRITE.
687 |  |
          *
688 |  |
          * @param key {@link SelectionKey}
689 |  |
          * @param ctx {@link Context}
690 |  |
          */
691 |  |
         public boolean onWriteInterest(final SelectionKey key,final Context ctx)
692 |  |
         throws IOException{
693 |  |
             // disable OP_WRITE on key before doing anything else
694 | 2241 |
             key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
695 |  |
696 |  |
             Object attach;
697 | 2241 |
             if (asyncQueueWriter.hasReadyAsyncWriteData(key)) {
698 | 2183 |
                 final Context context = pollContext(ctx, key, Context.OpType.OP_WRITE);
699 | 2183 |
                  invokeAsyncQueueWriter(context);
700 | 2183 |
                 return false;
701 | 58 |
             } else if ((attach = SelectionKeyAttachment.getAttachment(key))
702 |  |
                     instanceof CallbackHandler){
703 | 42 |
                 final Context context = pollContext(ctx, key, Context.OpType.OP_WRITE);
704 | 42 |
                 invokeCallbackHandler((CallbackHandler) attach, context);
705 | 42 |
                 return false;
706 |  |
             } else {
707 | 16 |
                 return true;
708 |  |
             }
709 |  |
         }
710 |  |
711 |  |
712 |  |
         /**
713 |  |
          * Handle OP_CONNECT.
714 |  |
          * @param key {@link SelectionKey}
715 |  |
          * @param ctx {@link Context}
716 |  |
          */
717 |  |
         public boolean onConnectInterest(final SelectionKey key, Context ctx)
718 |  |
         throws IOException{
719 |  |
             try {
720 |  |
                 // disable OP_CONNECT on key before doing anything else
721 | 205 |
                 key.interestOps(key.interestOps() & (~SelectionKey.OP_CONNECT));
722 |  |
723 |  |
                 // No OP_READ nor OP_WRITE allowed yet.
724 | 205 |
                 key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
725 | 205 |
                 key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
726 | 0 |
             } catch(CancelledKeyException e) {
727 |  |
                 // Even if key was cancelled - we need to notify CallBackHandler
728 | 0 |
                 if (logger.isLoggable(Level.FINE)) {
729 | 0 |
                     logger.log(Level.FINE, "CancelledKeyException occured when tried to change key interests", e);
730 |  |
                 }
731 | 205 |
             }
732 |  |
733 | 205 |
             Object attach = SelectionKeyAttachment.getAttachment(key);
734 | 205 |
             if (attach instanceof CallbackHandler){
735 | 205 |
                 final Context context = pollContext(ctx, key, Context.OpType.OP_CONNECT);
736 | 205 |
                 invokeCallbackHandler((CallbackHandler) attach, context);
737 |  |
             }
738 | 205 |
             return false;
739 |  |
         }
740 |  |
741 |  |
742 |  |
         /**
743 |  |
          * Invoke a CallbackHandler via a Context instance.
744 |  |
          * @param context {@link Context}
745 |  |
          * @throws java.io.IOException
746 |  |
          */
747 |  |
         protected void invokeCallbackHandler(CallbackHandler callbackHandler,
748 |  |
                 Context context) throws IOException {
749 | 58626 |
             IOEvent<Context>ioEvent = new IOEvent.DefaultIOEvent<Context>(context);
750 | 58626 |
             context.setIOEvent(ioEvent);
751 |  |
752 |  |
             // Added because of incompatibility with Grizzly 1.6.0
753 | 58626 |
             context.setSelectorHandler(this);
754 |  |
755 |  |
             try {
756 | 58626 |
                 CallbackHandlerContextTask task = CallbackHandlerContextTask.poll();
757 | 58626 |
                 task.setCallBackHandler(callbackHandler);
758 | 58626 |
                 boolean isRunInSeparateThread = true;
759 |  |
760 | 58626 |
                 if (callbackHandler instanceof CallbackHandlerDescriptor) {
761 | 0 |
                     isRunInSeparateThread =
762 |  |
                             ((CallbackHandlerDescriptor) callbackHandler).
763 |  |
                             isRunInSeparateThread(context.getCurrentOpType());
764 |  |
                 }
765 | 58626 |
                 context.execute(task, isRunInSeparateThread);
766 | 0 |
             } catch (PipelineFullException ex) {
767 | 0 |
                 throw new IOException(ex.getMessage());
768 | 58626 |
             }
769 | 58626 |
         }
770 |  |
771 |  |
772 |  |
         /**
773 |  |
          * Invoke a {@link AsyncQueueReader}
774 |  |
          * @param context {@link Context}
775 |  |
          * @throws java.io.IOException
776 |  |
          */
777 |  |
         protected void invokeAsyncQueueReader(Context context) throws IOException {
778 | 9757 |
             AsyncQueueReaderContextTask task = AsyncQueueReaderContextTask.poll();
779 | 9757 |
             task.setAsyncQueueReader(asyncQueueReader);
780 |  |
             try {
781 | 9757 |
                 context.execute(task);
782 | 0 |
             } catch (PipelineFullException ex) {
783 | 0 |
                 throw new IOException(ex.getMessage());
784 | 9757 |
             }
785 | 9757 |
         }
786 |  |
787 |  |
788 |  |
         /**
789 |  |
          * Invoke a {@link AsyncQueueWriter}
790 |  |
          * @param context {@link Context}
791 |  |
          * @throws java.io.IOException
792 |  |
          */
793 |  |
         protected void invokeAsyncQueueWriter(Context context) throws IOException {
794 | 2183 |
             AsyncQueueWriterContextTask task = AsyncQueueWriterContextTask.poll();
795 | 2183 |
             task.setAsyncQueueWriter(asyncQueueWriter);
796 |  |
             try {
797 | 2183 |
                 context.execute(task);
798 | 0 |
             } catch (PipelineFullException ex) {
799 | 0 |
                 throw new IOException(ex.getMessage());
800 | 2183 |
             }
801 | 2183 |
         }
802 |  |
803 |  |
804 |  |
         /**
805 |  |
          * Return an instance of the default {@link ConnectorHandler},
806 |  |
          * which is the {@link TCPConnectorHandler}
807 |  |
          * @return {@link ConnectorHandler}
808 |  |
          */
809 |  |
         public ConnectorHandler acquireConnectorHandler(){
810 | 175 |
             if (selector == null || !selector.isOpen()){
811 | 0 |
                 throw new IllegalStateException("SelectorHandler not yet started");
812 |  |
             }
813 |  |
814 | 175 |
             ConnectorHandler connectorHandler = connectorInstanceHandler.acquire();
815 | 175 |
             connectorHandler.setController(Controller.getHandlerController(this));
816 | 175 |
             return connectorHandler;
817 |  |
         }
818 |  |
819 |  |
820 |  |
         /**
821 |  |
          * Release a ConnectorHandler.
822 |  |
          */
823 |  |
         public void releaseConnectorHandler(ConnectorHandler connectorHandler){
824 | 174 |
             connectorInstanceHandler.release(connectorHandler);
825 | 174 |
         }
826 |  |
827 |  |
828 |  |
         /**
829 |  |
          * A token decribing the protocol supported by an implementation of this
830 |  |
          * interface
831 |  |
          */
832 |  |
         public Controller.Protocol protocol(){
833 | 49971 |
             return Controller.Protocol.TCP;
834 |  |
         }
835 |  |
         // ------------------------------------------------------ Utils ----------//
836 |  |
837 |  |
838 |  |
         /**
839 |  |
          * Initializes {@link SelectionKey} operation registries
840 |  |
          */
841 |  |
         protected void initOpRegistriesIfRequired() {
842 | 806205 |
             if (opToRegister == null){
843 | 135 |
                 opToRegister = new ConcurrentLinkedQueue<SelectionKeyOP>();
844 |  |
             }
845 | 806267 |
         }
846 |  |
847 |  |
         /**
848 |  |
          * {@inheritDoc}
849 |  |
          */
850 |  |
         public void configureChannel(SelectableChannel channel) throws IOException{
851 | 199 |
             Socket socket = ((SocketChannel) channel).socket();
852 |  |
853 | 199 |
             channel.configureBlocking(false);
854 |  |
855 | 199 |
             if (!channel.isOpen()){
856 | 0 |
                 return;
857 |  |
             }
858 |  |
859 |  |
             try{
860 | 199 |
                 if(socketTimeout >= 0 ) {
861 | 0 |
                     socket.setSoTimeout(socketTimeout);
862 |  |
                 }
863 | 0 |
             } catch (SocketException ex){
864 | 0 |
                 if (logger.isLoggable(Level.FINE)){
865 | 0 |
                     logger.log(Level.FINE,
866 |  |
                             "setSoTimeout exception ",ex);
867 |  |
                 }
868 | 199 |
             }
869 |  |
870 |  |
             try{
871 | 199 |
                 if(linger >= 0 ) {
872 | 0 |
                     socket.setSoLinger( true, linger);
873 |  |
                 }
874 | 0 |
             } catch (SocketException ex){
875 | 0 |
                 if (logger.isLoggable(Level.FINE)){
876 | 0 |
                     logger.log(Level.FINE,
877 |  |
                             "setSoLinger exception ",ex);
878 |  |
                 }
879 | 199 |
             }
880 |  |
881 |  |
             try{
882 | 199 |
                 socket.setTcpNoDelay(tcpNoDelay);
883 | 0 |
             } catch (SocketException ex){
884 | 0 |
                 if (logger.isLoggable(Level.FINE)){
885 | 0 |
                     logger.log(Level.FINE,
886 |  |
                             "setTcpNoDelay exception ",ex);
887 |  |
                 }
888 | 199 |
             }
889 |  |
890 |  |
             try{
891 | 199 |
                 socket.setReuseAddress(reuseAddress);
892 | 0 |
             } catch (SocketException ex){
893 | 0 |
                 if (logger.isLoggable(Level.FINE)){
894 | 0 |
                     logger.log(Level.FINE,
895 |  |
                             "setReuseAddress exception ",ex);
896 |  |
                 }
897 | 199 |
             }
898 | 199 |
         }
899 |  |
900 |  |
901 |  |
         // ------------------------------------------------------ Properties -----//
902 |  |
903 |  |
         public final Selector getSelector() {
904 | 867232 |
             return selector;
905 |  |
         }
906 |  |
907 |  |
         public final void setSelector(Selector selector) {
908 | 18 |
             this.selector = selector;
909 | 18 |
         }
910 |  |
911 |  |
         /**
912 |  |
          * {@inheritDoc}
913 |  |
          */
914 |  |
         public AsyncQueueReader getAsyncQueueReader() {
915 | 91104 |
             return asyncQueueReader;
916 |  |
         }
917 |  |
918 |  |
         /**
919 |  |
          * {@inheritDoc}
920 |  |
          */
921 |  |
         public AsyncQueueWriter getAsyncQueueWriter() {
922 | 910802 |
             return asyncQueueWriter;
923 |  |
         }
924 |  |
925 |  |
         public long getSelectTimeout() {
926 | 0 |
             return selectTimeout;
927 |  |
         }
928 |  |
929 |  |
         public void setSelectTimeout(long selectTimeout) {
930 | 0 |
             this.selectTimeout = selectTimeout;
931 | 0 |
         }
932 |  |
933 |  |
         public int getServerTimeout() {
934 | 0 |
             return serverTimeout;
935 |  |
         }
936 |  |
937 |  |
         public void setServerTimeout(int serverTimeout) {
938 | 0 |
             this.serverTimeout = serverTimeout;
939 | 0 |
         }
940 |  |
941 |  |
         public InetAddress getInet() {
942 | 0 |
             return inet;
943 |  |
         }
944 |  |
945 |  |
         public void setInet(InetAddress inet) {
946 | 0 |
             this.inet = inet;
947 | 0 |
         }
948 |  |
949 |  |
         /**
950 |  |
          * Gets this {@link SelectorHandler} current role.
951 |  |
          * <tt>TCPSelectorHandler</tt> could act as client, which corresponds to
952 |  |
          * {@link Role#CLIENT} or client-server, which corresponds
953 |  |
          * to the {@link Role#CLIENT_SERVER}
954 |  |
          *
955 |  |
          * @return the {@link Role}
956 |  |
          */
957 |  |
         public Role getRole() {
958 | 0 |
             return role;
959 |  |
         }
960 |  |
961 |  |
         /**
962 |  |
          * Sets this {@link SelectorHandler} current role.
963 |  |
          * <tt>TCPSelectorHandler</tt> could act as client, which corresponds to
964 |  |
          * {@link Role#CLIENT} or client-server, which corresponds
965 |  |
          * to the {@link Role#CLIENT_SERVER}
966 |  |
          *
967 |  |
          * @param role the {@link Role}
968 |  |
          */
969 |  |
         public void setRole(Role role) {
970 | 0 |
             this.role = role;
971 | 0 |
         }
972 |  |
973 |  |
         /**
974 |  |
          * Returns port number {@link SelectorHandler} is listening on
975 |  |
          * Similar to <code>getPort()</code>, but getting port number directly from
976 |  |
          * connection ({@link ServerSocket}, {@link DatagramSocket}).
977 |  |
          * So if default port number 0 was set during initialization, then <code>getPort()</code>
978 |  |
          * will return 0, but getPortLowLevel() will
979 |  |
          * return port number assigned by OS.
980 |  |
          *
981 |  |
          * @return port number or -1 if {@link SelectorHandler} was not initialized for accepting connections.
982 |  |
          */
983 |  |
         public int getPortLowLevel() {
984 | 2 |
             if (serverSocket != null) {
985 | 2 |
                 return serverSocket.getLocalPort();
986 |  |
             }
987 |  |
988 | 0 |
             return -1;
989 |  |
         }
990 |  |
991 |  |
         public int getPort() {
992 | 3 |
             return port;
993 |  |
         }
994 |  |
995 |  |
         public void setPort(int port) {
996 | 39 |
             this.port = port;
997 | 39 |
         }
998 |  |
999 |  |
         public int getSsBackLog() {
1000 | 0 |
             return ssBackLog;
1001 |  |
         }
1002 |  |
1003 |  |
         public void setSsBackLog(int ssBackLog) {
1004 | 0 |
             this.ssBackLog = ssBackLog;
1005 | 0 |
         }
1006 |  |
1007 |  |
1008 |  |
         /**
1009 |  |
          * Return the tcpNoDelay value used by the underlying accepted Sockets.
1010 |  |
          *
1011 |  |
          * Also see setTcpNoDelay(boolean tcpNoDelay)
1012 |  |
          */
1013 |  |
         public boolean isTcpNoDelay() {
1014 | 0 |
             return tcpNoDelay;
1015 |  |
         }
1016 |  |
1017 |  |
1018 |  |
         /**
1019 |  |
          * Enable (true) or disable (false) the underlying Socket's
1020 |  |
          * tcpNoDelay.
1021 |  |
          *
1022 |  |
          * Default value for tcpNoDelay is enabled (set to true), as according to
1023 |  |
          * the performance tests, it performs better for most cases.
1024 |  |
          *
1025 |  |
          * The Connector side should also set tcpNoDelay the same as it is set here
1026 |  |
          * whenever possible.
1027 |  |
          */
1028 |  |
         public void setTcpNoDelay(boolean tcpNoDelay) {
1029 | 0 |
             this.tcpNoDelay = tcpNoDelay;
1030 | 0 |
         }
1031 |  |
1032 |  |
         public int getLinger() {
1033 | 0 |
             return linger;
1034 |  |
         }
1035 |  |
1036 |  |
         public void setLinger(int linger) {
1037 | 0 |
             this.linger = linger;
1038 | 0 |
         }
1039 |  |
1040 |  |
         public int getSocketTimeout() {
1041 | 0 |
             return socketTimeout;
1042 |  |
         }
1043 |  |
1044 |  |
         public void setSocketTimeout(int socketTimeout) {
1045 | 0 |
             this.socketTimeout = socketTimeout;
1046 | 0 |
         }
1047 |  |
1048 |  |
         public Logger getLogger() {
1049 | 0 |
             return logger;
1050 |  |
         }
1051 |  |
1052 |  |
         public void setLogger(Logger logger) {
1053 | 0 |
             this.logger = logger;
1054 | 0 |
         }
1055 |  |
1056 |  |
         public boolean isReuseAddress() {
1057 | 0 |
             return reuseAddress;
1058 |  |
         }
1059 |  |
1060 |  |
         public void setReuseAddress(boolean reuseAddress) {
1061 | 0 |
             this.reuseAddress = reuseAddress;
1062 | 0 |
         }
1063 |  |
1064 |  |
         /**
1065 |  |
          * {@inheritDoc}
1066 |  |
          */
1067 |  |
         public Pipeline pipeline(){
1068 | 90804 |
             return pipeline;
1069 |  |
         }
1070 |  |
1071 |  |
1072 |  |
         /**
1073 |  |
          * {@inheritDoc}
1074 |  |
          */
1075 |  |
         public void setPipeline(Pipeline pipeline){
1076 | 0 |
             this.pipeline = pipeline;
1077 | 0 |
         }
1078 |  |
1079 |  |
1080 |  |
         /**
1081 |  |
          * {@inheritDoc}
1082 |  |
          */
1083 |  |
         public Class<? extends SelectionKeyHandler> getPreferredSelectionKeyHandler() {
1084 | 238 |
             return DefaultSelectionKeyHandler.class;
1085 |  |
         }
1086 |  |
1087 |  |
1088 |  |
         /**
1089 |  |
          * Get the SelectionKeyHandler associated with this SelectorHandler.
1090 |  |
          */
1091 |  |
         public SelectionKeyHandler getSelectionKeyHandler() {
1092 | 565487 |
             return selectionKeyHandler;
1093 |  |
         }
1094 |  |
1095 |  |
1096 |  |
         /**
1097 |  |
          * Set SelectionKeyHandler associated with this SelectorHandler.
1098 |  |
          */
1099 |  |
         public void setSelectionKeyHandler(SelectionKeyHandler selectionKeyHandler) {
1100 | 135 |
             this.selectionKeyHandler = selectionKeyHandler;
1101 | 135 |
             this.selectionKeyHandler.setSelectorHandler(this);
1102 | 135 |
         }
1103 |  |
1104 |  |
1105 |  |
         /**
1106 |  |
          * Set the {@link ProtocolChainInstanceHandler} to use for
1107 |  |
          * creating instance of {@link ProtocolChain}.
1108 |  |
          */
1109 |  |
         public void setProtocolChainInstanceHandler(ProtocolChainInstanceHandler
1110 |  |
                 instanceHandler){
1111 | 0 |
             this.instanceHandler = instanceHandler;
1112 | 0 |
         }
1113 |  |
1114 |  |
1115 |  |
         /**
1116 |  |
          * Return the {@link ProtocolChainInstanceHandler}
1117 |  |
          */
1118 |  |
         public ProtocolChainInstanceHandler getProtocolChainInstanceHandler(){
1119 | 252174 |
             return instanceHandler;
1120 |  |
         }
1121 |  |
1122 |  |
         /**
1123 |  |
          * {@inheritDoc}
1124 |  |
          */
1125 |  |
         public void closeChannel(SelectableChannel channel) {
1126 |  |
             // channel could be either SocketChannel or ServerSocketChannel
1127 | 408 |
             if (channel instanceof SocketChannel) {
1128 | 374 |
                 Socket socket = ((SocketChannel) channel).socket();
1129 |  |
1130 |  |
                 try {
1131 | 374 |
                     if (!socket.isInputShutdown()) socket.shutdownInput();
1132 | 0 |
                 } catch (IOException ex){
1133 |  |
                     ;
1134 | 374 |
                 }
1135 |  |
1136 |  |
                 try {
1137 | 374 |
                     if (!socket.isOutputShutdown()) socket.shutdownOutput();
1138 | 0 |
                 } catch (IOException ex){
1139 |  |
                     ;
1140 | 374 |
                 }
1141 |  |
1142 |  |
                 try{
1143 | 374 |
                     socket.close();
1144 | 0 |
                 } catch (IOException ex){
1145 |  |
                     ;
1146 | 374 |
                 }
1147 |  |
             }
1148 |  |
1149 |  |
             try{
1150 | 408 |
                 channel.close();
1151 | 0 |
             } catch (IOException ex){
1152 |  |
                 ; // LOG ME
1153 | 408 |
             }
1154 |  |
1155 | 408 |
             if (asyncQueueReader != null) {
1156 | 408 |
                 asyncQueueReader.onClose(channel);
1157 |  |
             }
1158 |  |
1159 | 408 |
             if (asyncQueueWriter != null) {
1160 | 408 |
                 asyncQueueWriter.onClose(channel);
1161 |  |
             }
1162 | 408 |
         }
1163 |  |
1164 |  |
         /**
1165 |  |
          * Polls {@link Context} from pool and initializes it.
1166 |  |
          *
1167 |  |
          * @param serverContext {@link Controller} context
1168 |  |
          * @param key {@link SelectionKey}
1169 |  |
          * @return {@link Context}
1170 |  |
          */
1171 |  |
         protected Context pollContext(final Context serverContext,
1172 |  |
                 final SelectionKey key, final Context.OpType opType) {
1173 | 70566 |
             ProtocolChain protocolChain = instanceHandler != null ?
1174 |  |
                 instanceHandler.poll() :
1175 |  |
                 serverContext.getController().getProtocolChainInstanceHandler().poll();
1176 |  |
1177 | 70566 |
             final Context context = serverContext.getController().pollContext(key, opType);
1178 | 70566 |
             context.setSelectionKey(key);
1179 | 70566 |
             context.setSelectorHandler(this);
1180 | 70566 |
             context.setAsyncQueueReader(asyncQueueReader);
1181 | 70566 |
             context.setAsyncQueueWriter(asyncQueueWriter);
1182 | 70566 |
             context.setProtocolChain(protocolChain);
1183 | 70566 |
             return context;
1184 |  |
         }
1185 |  |
1186 |  |
         //--------------- ConnectorInstanceHandler -----------------------------
1187 |  |
         /**
1188 |  |
          * Return <Callable>factory<Callable> object, which knows how
1189 |  |
          * to create {@link ConnectorInstanceHandler} corresponding to the protocol
1190 |  |
          * @return <Callable>factory</code>
1191 |  |
          */
1192 |  |
         protected Callable<ConnectorHandler> getConnectorInstanceHandlerDelegate() {
1193 | 103 |
             return new Callable<ConnectorHandler>() {
1194 |  |
                 public ConnectorHandler call() throws Exception {
1195 | 5 |
                     return new TCPConnectorHandler();
1196 |  |
                 }
1197 |  |
             };
1198 |  |
         }
1199 |  |
1200 |  |
         // ----------- AttributeHolder interface implementation ----------- //
1201 |  |
1202 |  |
         /**
1203 |  |
          * Remove a key/value object.
1204 |  |
          * Method is not thread safe
1205 |  |
          *
1206 |  |
          * @param key - name of an attribute
1207 |  |
          * @return  attribute which has been removed
1208 |  |
          */
1209 |  |
         public Object removeAttribute(String key) {
1210 | 0 |
             if (attributes == null) return null;
1211 |  |
1212 | 0 |
             return attributes.remove(key);
1213 |  |
         }
1214 |  |
1215 |  |
         /**
1216 |  |
          * Set a key/value object.
1217 |  |
          * Method is not thread safe
1218 |  |
          *
1219 |  |
          * @param key - name of an attribute
1220 |  |
          * @param value - value of named attribute
1221 |  |
          */
1222 |  |
         public void setAttribute(String key, Object value) {
1223 | 0 |
             if (attributes == null) {
1224 | 0 |
                 attributes = new HashMap<String, Object>();
1225 |  |
             }
1226 |  |
1227 | 0 |
             attributes.put(key, value);
1228 | 0 |
         }
1229 |  |
1230 |  |
         /**
1231 |  |
          * Return an object based on a key.
1232 |  |
          * Method is not thread safe
1233 |  |
          *
1234 |  |
          * @param key - name of an attribute
1235 |  |
          * @return - attribute value for the <tt>key</tt>, null if <tt>key</tt>
1236 |  |
          *           does not exist in <tt>attributes</tt>
1237 |  |
          */
1238 |  |
         public Object getAttribute(String key) {
1239 | 0 |
             if (attributes == null) return null;
1240 |  |
1241 | 0 |
             return attributes.get(key);
1242 |  |
         }
1243 |  |
1244 |  |
1245 |  |
         /**
1246 |  |
          * Set a {@link Map} of attribute name/value pairs.
1247 |  |
          * Old {@link AttributeHolder} values will not be available.
1248 |  |
          * Later changes of this {@link Map} will lead to changes to the current
1249 |  |
          * {@link AttributeHolder}.
1250 |  |
          *
1251 |  |
          * @param attributes - map of name/value pairs
1252 |  |
          */
1253 |  |
         public void setAttributes(Map<String, Object> attributes) {
1254 | 6 |
             this.attributes = attributes;
1255 | 6 |
         }
1256 |  |
1257 |  |
1258 |  |
         /**
1259 |  |
          * Return a {@link Map} of attribute name/value pairs.
1260 |  |
          * Updates, performed on the returned {@link Map} will be reflected in
1261 |  |
          * this {@link AttributeHolder}
1262 |  |
          *
1263 |  |
          * @return - {@link Map} of attribute name/value pairs
1264 |  |
          */
1265 |  |
         public Map<String, Object> getAttributes() {
1266 | 6 |
             return attributes;
1267 |  |
         }
1268 |  |
1269 |  |
         /**
1270 |  |
          * Returns the {@link Role}, depending on isClient value
1271 |  |
          * @param isClient <tt>true>tt>, if this <tt>SelectorHandler</tt> works in
1272 |  |
          *          the client mode, or <tt>false</tt> otherwise.
1273 |  |
          * @return {@link Role}
1274 |  |
          */
1275 |  |
         protected static Role boolean2Role(boolean isClient) {
1276 | 79 |
             if (isClient) return Role.CLIENT;
1277 |  |
1278 | 0 |
             return Role.CLIENT_SERVER;
1279 |  |
         }
1280 |  |
     }