##### Coverage Report - com.sun.grizzly.filter.ReadFilter

Classes in this File | Line Coverage | Branch Coverage | Complexity
ReadFilter |
73 % | 58/79
55 % | 31/56
| 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.filter;
40 |  |
41 |  |
     import com.sun.grizzly.Context;
42 |  |
     import com.sun.grizzly.Controller;
43 |  |
     import com.sun.grizzly.ProtocolFilter;
44 |  |
     import com.sun.grizzly.util.WorkerThread;
45 |  |
     import java.net.SocketAddress;
46 |  |
     import java.io.IOException;
47 |  |
     import java.nio.ByteBuffer;
48 |  |
     import java.nio.channels.DatagramChannel;
49 |  |
     import java.nio.channels.SelectionKey;
50 |  |
     import java.nio.channels.SocketChannel;
51 |  |
     import java.util.logging.Level;
52 |  |
53 |  |
     import static com.sun.grizzly.Controller.Protocol.TCP;
54 |  |
     import static com.sun.grizzly.Controller.Protocol.TLS;
55 |  |
     import static com.sun.grizzly.Controller.Protocol.UDP;
56 |  |
     import com.sun.grizzly.Controller.Protocol;
57 |  |
     import com.sun.grizzly.ProtocolChain;
58 |  |
     import com.sun.grizzly.ReinvokeAware;
59 |  |
     import com.sun.grizzly.SelectorHandler;
60 |  |
61 |  |
     /**
62 |  |
      * Simple {@link ProtocolFilter} implementation which read the available bytes
63 |  |
      * and delegate the processing to the next {@link ProtocolFilter} in the {@link ProtocolChain}.
64 |  |
      * If no bytes are available, no new {@link ProtocolFilter} will be a invoked and
65 |  |
      * the connection (SelectionKey) will be cancelled. This filter can be used
66 |  |
      * for both UDP (reveive) and TCP (read).
67 |  |
      *
68 |  |
      * Note that all ready OP_WRITE operations will be ignored.
69 |  |
      *
70 |  |
      * @author Jeanfrancois Arcand
71 |  |
      */
72 |  |
     public class ReadFilter implements ProtocolFilter{
73 |  |
74 |  |
         public final static String UDP_SOCKETADDRESS = "socketAddress";
75 |  |
76 |  |
77 |  |
         /**
78 |  |
          * <tt>true</tt> if a pipelined execution is required. A pipelined execution
79 |  |
          * occurs when a ProtocolFilter implementation set the
80 |  |
          * ProtocolFilter.READ_SUCCESS as an attribute to a Context. When this
81 |  |
          * attribute is present, the ProtocolChain will not release the current
82 |  |
          * running Thread and will re-execute all its ProtocolFilter.
83 |  |
          */
84 | 33 |
         protected boolean continousExecution = false;
85 |  |
86 | 33 |
         protected int readAttempts = 3;
87 |  |
88 | 33 |
         public ReadFilter(){
89 | 33 |
         }
90 |  |
91 |  |
         /**
92 |  |
          * Read available bytes and delegate the processing of them to the next
93 |  |
          * {@link ProtocolFilter} in the {@link ProtocolChain}.
94 |  |
          * @return <tt>true</tt> if the next ProtocolFilter on the ProtocolChain
95 |  |
          *                       need to bve invoked.
96 |  |
          */
97 |  |
         public boolean execute(Context ctx) throws IOException {
98 | 37600 |
             return execute(ctx, null);
99 |  |
         }
100 |  |
101 |  |
102 |  |
         /**
103 |  |
          * Read available bytes to the specific {@link ByteBuffer} and delegate
104 |  |
          * the processing of them to the next ProtocolFilter in the ProtocolChain.
105 |  |
          * @return <tt>true</tt> if the next ProtocolFilter on the ProtocolChain
106 |  |
          *                       need to bve invoked.
107 |  |
          */
108 |  |
         protected boolean execute(Context ctx, ByteBuffer byteBuffer) throws IOException {
109 |  |
110 | 37600 |
             if (ctx.getCurrentOpType() == Context.OpType.OP_WRITE){
111 | 2 |
                 if (Controller.logger().isLoggable(Level.FINE)){
112 | 0 |
                     Controller.logger().fine("ReadFilter cannont handle OP_WRITE");
113 |  |
                 }
114 | 2 |
                 return false;
115 |  |
             }
116 |  |
117 |  |
118 | 37598 |
             if (byteBuffer == null) {
119 | 37598 |
                 byteBuffer = ((WorkerThread)Thread.currentThread()).getByteBuffer();
120 |  |
             }
121 |  |
122 | 37598 |
             if (!byteBuffer.hasRemaining()){
123 | 0 |
                 throw new IllegalStateException("ByteBuffer is full: " + byteBuffer);
124 |  |
             }
125 |  |
126 |  |
127 | 37598 |
             boolean invokeNextFilter = true;
128 | 37598 |
             int count = -1;
129 | 37598 |
             SocketAddress socketAddress = null;
130 | 37598 |
             Exception exception = null;
131 | 37598 |
             SelectionKey key = ctx.getSelectionKey();
132 |  |
133 | 37598 |
             Protocol protocol = ctx.getProtocol();
134 |  |
             try {
135 | 37598 |
                 int loop = 0;
136 | 37598 |
                 if (protocol == TCP || protocol == TLS){
137 | 16596 |
                     SocketChannel channel = (SocketChannel)key.channel();
138 |  |
139 |  |
                     // As soon as bytes are ready, invoke the next ProtocolFilter.
140 | 16596 |
                     while ((count = channel.read(byteBuffer)) == 0) {
141 |  |
142 |  |
                         // Avoid calling the Selector.
143 | 0 |
                         if (++loop >= readAttempts){
144 | 0 |
                             if (ctx.getKeyRegistrationState()
145 |  |
                                     != Context.KeyRegistrationState.NONE){
146 | 0 |
                                 ctx.setAttribute(ProtocolFilter.SUCCESSFUL_READ,
147 |  |
                                                  Boolean.FALSE);
148 | 0 |
                                 invokeNextFilter = false;
149 |  |
                             }
150 |  |
                             break;
151 |  |
                         }
152 |  |
                     }
153 | 16595 |
                 } else if (protocol == UDP){
154 | 21002 |
                     DatagramChannel datagramChannel = (DatagramChannel)key.channel();
155 | 21002 |
                     socketAddress = datagramChannel.receive(byteBuffer);
156 | 21002 |
                     ctx.getSelectorHandler().register(key, SelectionKey.OP_READ);
157 |  |
                 }
158 | 1 |
             } catch (IOException ex) {
159 | 1 |
                 exception = ex;
160 | 1 |
                 log("ReadFilter.execute",ex);
161 | 0 |
             } catch (RuntimeException ex) {
162 | 0 |
                 exception = ex;
163 | 0 |
                 log("ReadFilter.execute",ex);
164 |  |
             } finally {
165 | 37598 |
                 if (exception != null){
166 | 1 |
                     ctx.setAttribute(Context.THROWABLE,exception);
167 | 1 |
                     if (protocol != UDP){
168 | 1 |
                         ctx.setKeyRegistrationState(
169 |  |
                             Context.KeyRegistrationState.CANCEL);
170 |  |
                     }
171 | 1 |
                     invokeNextFilter = false;
172 | 37597 |
                 } else if (count == -1 && protocol != UDP){
173 | 42 |
                     ctx.setKeyRegistrationState(
174 |  |
                             Context.KeyRegistrationState.CANCEL);
175 | 42 |
                     invokeNextFilter = false;
176 | 37555 |
                 } else if (socketAddress == null && protocol == UDP ){
177 | 0 |
                     ctx.setKeyRegistrationState(Context.KeyRegistrationState.REGISTER);
178 | 0 |
                     invokeNextFilter = false;
179 | 37555 |
                 } else if (protocol == UDP) {
180 | 21002 |
                     ctx.setAttribute(UDP_SOCKETADDRESS,socketAddress);
181 |  |
                 }
182 |  |
             }
183 | 37598 |
             return invokeNextFilter;
184 |  |
         }
185 |  |
186 |  |
187 |  |
         /**
188 |  |
          * If no bytes were available, close the connection by cancelling the
189 |  |
          * SelectionKey. If bytes were available, register the SelectionKey
190 |  |
          * for new bytes.
191 |  |
          *
192 |  |
          * @return <tt>true</tt> if the previous ProtocolFilter postExecute method
193 |  |
          *         needs to be invoked.
194 |  |
          */
195 |  |
         public boolean postExecute(Context ctx) throws IOException {
196 |  |
197 | 37597 |
             final SelectorHandler selectorHandler =
198 |  |
                     ctx.getSelectorHandler();
199 | 37597 |
             final SelectionKey key = ctx.getSelectionKey();
200 | 37597 |
             final Context.KeyRegistrationState state = ctx.getKeyRegistrationState();
201 | 37597 |
             final Protocol protocol = ctx.getProtocol();
202 |  |
203 |  |
             try{
204 |  |
                 //For UDP, we don't have to do anything as the OP_READ operations
205 |  |
                 //as already been handled, and cencelling the key is not allowed.
206 | 37597 |
                 if (protocol == UDP){
207 | 21004 |
                     return true;
208 |  |
                 }
209 |  |
210 |  |
211 | 16593 |
                 ProtocolChain protocolChain = ctx.getProtocolChain();
212 |  |
213 |  |
                 // Check if both Filter and ProtocolChain are
214 |  |
                 // set to reinvoke the protocol chain
215 | 16593 |
                 boolean isReinvoke = continousExecution &&
216 |  |
                         (protocolChain instanceof ReinvokeAware) &&
217 |  |
                         ((ReinvokeAware) protocolChain).isContinuousExecution();
218 |  |
219 |  |
                 // The ProtocolChain associated with this ProtocolFilter will re-invoke
220 |  |
                 // the execute method. Do not register the SelectionKey in that case
221 |  |
                 // to avoid thread races.
222 | 16593 |
                 if (isReinvoke
223 |  |
                         && state == Context.KeyRegistrationState.REGISTER
224 |  |
                         && Boolean.FALSE !=
225 |  |
                             (Boolean)ctx.getAttribute(ProtocolFilter.SUCCESSFUL_READ)){
226 | 0 |
                     ctx.setAttribute(ProtocolFilter.SUCCESSFUL_READ,
227 |  |
                                      Boolean.TRUE);
228 |  |
                 } else {
229 | 16593 |
                     if (state == Context.KeyRegistrationState.CANCEL){
230 | 43 |
                         selectorHandler.getSelectionKeyHandler().cancel(key);
231 | 16550 |
                     } else if (state == Context.KeyRegistrationState.REGISTER){
232 | 16546 |
                         selectorHandler.register(key, SelectionKey.OP_READ);
233 |  |
                     }
234 |  |
                 }
235 | 16593 |
                 return true;
236 |  |
             } finally {
237 | 37597 |
                 ctx.removeAttribute(Context.THROWABLE);
238 | 37597 |
                 ctx.removeAttribute(UDP_SOCKETADDRESS);
239 |  |
             }
240 |  |
         }
241 |  |
242 |  |
243 |  |
         /**
244 |  |
          * Set to <tt>true</tt> if the current {@link Pipeline} can
245 |  |
          * re-execute its ProtocolFilter(s) after a successful execution. Enabling
246 |  |
          * this property is useful for protocol that needs to support pipelined
247 |  |
          * message requests as the ProtocolFilter are automatically re-executed,
248 |  |
          * avoiding the overhead of releasing the current Thread, registering
249 |  |
          * back the SelectionKey to the {@link SelectorHandler} and waiting for a new
250 |  |
          * NIO event.
251 |  |
          *
252 |  |
          * Some protocols (like http) can get the http headers in one
253 |  |
          * SocketChannel.read, parse the message and then get the next http message
254 |  |
          * on the second SocketChannel.read(). Not having to release the Thread
255 |  |
          * and re-execute the ProtocolFilter greatly improve performance.
256 |  |
          * @param continousExecution true to enable continuous execution.
257 |  |
          *        (default is false).
258 |  |
          */
259 |  |
         public void setContinuousExecution(boolean continousExecution){
260 | 0 |
             this.continousExecution = continousExecution;
261 | 0 |
         }
262 |  |
263 |  |
264 |  |
         /**
265 |  |
          * Return <tt>true</tt> if the current {@link Pipeline} can
266 |  |
          * re-execute its ProtocolFilter after a successful execution.
267 |  |
          */
268 |  |
         public boolean isContinuousExecution(){
269 | 0 |
             return continousExecution;
270 |  |
         }
271 |  |
272 |  |
         /**
273 |  |
          * Get the number of attempts the {@link ReadFilter} will try to read a data
274 |  |
          * from a channel.
275 |  |
          *
276 |  |
          * @return the number of attempts the {@link ReadFilter} will try to read a data
277 |  |
          * from a channel.
278 |  |
          */
279 |  |
         public int getReadAttempts() {
280 | 0 |
             return readAttempts;
281 |  |
         }
282 |  |
283 |  |
         /**
284 |  |
          * Set the number of attempts the {@link ReadFilter} will try to read a data
285 |  |
          * from a channel.
286 |  |
          *
287 |  |
          * @param readAttempts the number of attempts the {@link ReadFilter} will
288 |  |
          * try to read a data from a channel.
289 |  |
          */
290 |  |
         public void setReadAttempts(int readAttempts) {
291 | 0 |
             if (readAttempts < 1) {
292 | 0 |
                 throw new IllegalArgumentException("The readAttempts parameter should be >= 1");
293 |  |
             }
294 |  |
295 | 0 |
             this.readAttempts = readAttempts;
296 | 0 |
         }
297 |  |
298 |  |
299 |  |
         /**
300 |  |
          * Log a message/exception.
301 |  |
          * @param msg <code>String</code>
302 |  |
          * @param t <code>Throwable</code>
303 |  |
          */
304 |  |
         protected void log(String msg,Throwable t){
305 | 1 |
             if (Controller.logger().isLoggable(Level.FINE)){
306 | 0 |
                 Controller.logger().log(Level.FINE, msg, t);
307 |  |
             }
308 | 1 |
         }
309 |  |
     }