View Javadoc
1   /*
2    *  Licensed to the Apache Software Foundation (ASF) under one
3    *  or more contributor license agreements.  See the NOTICE file
4    *  distributed with this work for additional information
5    *  regarding copyright ownership.  The ASF licenses this file
6    *  to you under the Apache License, Version 2.0 (the
7    *  "License"); you may not use this file except in compliance
8    *  with the License.  You may obtain a copy of the License at
9    *
10   *    http://www.apache.org/licenses/LICENSE-2.0
11   *
12   *  Unless required by applicable law or agreed to in writing,
13   *  software distributed under the License is distributed on an
14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   *  KIND, either express or implied.  See the License for the
16   *  specific language governing permissions and limitations
17   *  under the License.
18   *
19   */
20  package org.apache.mina.filter.codec;
21  
22  import org.apache.mina.core.buffer.IoBuffer;
23  import org.apache.mina.core.service.TransportMetadata;
24  import org.apache.mina.core.session.AttributeKey;
25  import org.apache.mina.core.session.IoSession;
26  
27  /**
28   * A {@link ProtocolDecoder} that cumulates the content of received buffers to a
29   * <em>cumulative buffer</em> to help users implement decoders.
30   * <p>
31   * If the received {@link IoBuffer} is only a part of a message. decoders should
32   * cumulate received buffers to make a message complete or to postpone decoding
33   * until more buffers arrive.
34   * <p>
35   * Here is an example decoder that decodes CRLF terminated lines into
36   * <code>Command</code> objects:
37   * 
38   * <pre>
39   * public class CrLfTerminatedCommandLineDecoder extends CumulativeProtocolDecoder {
40   * 
41   *     private Command parseCommand(IoBuffer in) {
42   *         // Convert the bytes in the specified buffer to a
43   *         // Command object.
44   *         ...
45   *     }
46   * 
47   *     protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
48   * 
49   *         // Remember the initial position.
50   *         int start = in.position();
51   * 
52   *         // Now find the first CRLF in the buffer.
53   *         byte previous = 0;
54   *         while (in.hasRemaining()) {
55   *             byte current = in.get();
56   * 
57   *             if (previous == '\r' &amp;&amp; current == '\n') {
58   *                 // Remember the current position and limit.
59   *                 int position = in.position();
60   *                 int limit = in.limit();
61   *                 try {
62   *                     in.position(start);
63   *                     in.limit(position);
64   *                     // The bytes between in.position() and in.limit()
65   *                     // now contain a full CRLF terminated line.
66   *                     out.write(parseCommand(in.slice()));
67   *                 } finally {
68   *                     // Set the position to point right after the
69   *                     // detected line and set the limit to the old
70   *                     // one.
71   *                     in.position(position);
72   *                     in.limit(limit);
73   *                 }
74   *                 // Decoded one line; CumulativeProtocolDecoder will
75   *                 // call me again until I return false. So just
76   *                 // return true until there are no more lines in the
77   *                 // buffer.
78   *                 return true;
79   *             }
80   * 
81   *             previous = current;
82   *         }
83   * 
84   *         // Could not find CRLF in the buffer. Reset the initial
85   *         // position to the one we recorded above.
86   *         in.position(start);
87   * 
88   *         return false;
89   *     }
90   * }
91   * </pre>
92   * <p>
93   * Please note that this decoder simply forward the call to
94   * doDecode(IoSession, IoBuffer, ProtocolDecoderOutput) if the
95   * underlying transport doesn't have a packet fragmentation. Whether the
96   * transport has fragmentation or not is determined by querying
97   * {@link TransportMetadata}.
98   *
99   * @author <a href="http://mina.apache.org">Apache MINA Project</a>
100  */
101 public abstract class CumulativeProtocolDecoder extends ProtocolDecoderAdapter {
102     /** The buffer used to store the data in the session */
103     private static final AttributeKey BUFFER = new AttributeKey(CumulativeProtocolDecoder.class, "buffer");
104     
105     /** A flag set to true if we handle fragmentation accordingly to the TransportMetadata setting. 
106      * It can be set to false if needed (UDP with fragments, for instance). the default value is 'true'
107      */
108     private boolean transportMetadataFragmentation = true;
109 
110     /**
111      * Creates a new instance.
112      */
113     protected CumulativeProtocolDecoder() {
114         // Do nothing
115     }
116 
117     /**
118      * Cumulates content of <tt>in</tt> into internal buffer and forwards
119      * decoding request to
120      * doDecode(IoSession, IoBuffer, ProtocolDecoderOutput).
121      * <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
122      * and the cumulative buffer is compacted after decoding ends.
123      *
124      * @throws IllegalStateException
125      *             if your <tt>doDecode()</tt> returned <tt>true</tt> not
126      *             consuming the cumulative buffer.
127      */
128     @Override
129     public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
130         if (transportMetadataFragmentation && !session.getTransportMetadata().hasFragmentation()) {
131             while (in.hasRemaining()) {
132                 if (!doDecode(session, in, out)) {
133                     break;
134                 }
135             }
136 
137             return;
138         }
139 
140         boolean usingSessionBuffer = true;
141         IoBuffer buf = (IoBuffer) session.getAttribute(BUFFER);
142         // If we have a session buffer, append data to that; otherwise
143         // use the buffer read from the network directly.
144         if (buf != null) {
145             boolean appended = false;
146             // Make sure that the buffer is auto-expanded.
147             if (buf.isAutoExpand()) {
148                 try {
149                     buf.put(in);
150                     appended = true;
151                 } catch (IllegalStateException | IndexOutOfBoundsException e) {
152                     // A user called derivation method (e.g. slice()),
153                     // which disables auto-expansion of the parent buffer.
154                 }
155             }
156 
157             if (appended) {
158                 buf.flip();
159             } else {
160                 // Reallocate the buffer if append operation failed due to
161                 // derivation or disabled auto-expansion.
162                 buf.flip();
163                 IoBuffer newBuf = IoBuffer.allocate(buf.remaining() + in.remaining()).setAutoExpand(true);
164                 newBuf.order(buf.order());
165                 newBuf.put(buf);
166                 newBuf.put(in);
167                 newBuf.flip();
168                 buf = newBuf;
169 
170                 // Update the session attribute.
171                 session.setAttribute(BUFFER, buf);
172             }
173         } else {
174             buf = in;
175             usingSessionBuffer = false;
176         }
177 
178         for (;;) {
179             int oldPos = buf.position();
180             boolean decoded = doDecode(session, buf, out);
181             if (decoded) {
182                 if (buf.position() == oldPos) {
183                     throw new IllegalStateException("doDecode() can't return true when buffer is not consumed.");
184                 }
185 
186                 if (!buf.hasRemaining()) {
187                     break;
188                 }
189             } else {
190                 break;
191             }
192         }
193 
194         // if there is any data left that cannot be decoded, we store
195         // it in a buffer in the session and next time this decoder is
196         // invoked the session buffer gets appended to
197         if (buf.hasRemaining()) {
198             if (usingSessionBuffer && buf.isAutoExpand()) {
199                 buf.compact();
200             } else {
201                 storeRemainingInSession(buf, session);
202             }
203         } else {
204             if (usingSessionBuffer) {
205                 removeSessionBuffer(session);
206             }
207         }
208     }
209 
210     /**
211      * Implement this method to consume the specified cumulative buffer and
212      * decode its content into message(s).
213      *
214      * @param session The current Session
215      * @param in the cumulative buffer
216      * @param out The {@link ProtocolDecoderOutput} that will receive the decoded message
217      * @return <tt>true</tt> if and only if there's more to decode in the buffer
218      *         and you want to have <tt>doDecode</tt> method invoked again.
219      *         Return <tt>false</tt> if remaining data is not enough to decode,
220      *         then this method will be invoked again when more data is
221      *         cumulated.
222      * @throws Exception if cannot decode <tt>in</tt>.
223      */
224     protected abstract boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception;
225 
226     /**
227      * Releases the cumulative buffer used by the specified <tt>session</tt>.
228      * Please don't forget to call <tt>super.dispose( session )</tt> when you
229      * override this method.
230      */
231     @Override
232     public void dispose(IoSession session) throws Exception {
233         removeSessionBuffer(session);
234     }
235 
236     private void removeSessionBuffer(IoSession session) {
237         session.removeAttribute(BUFFER);
238     }
239 
240     private void storeRemainingInSession(IoBuffer buf, IoSession session) {
241         final IoBuffer remainingBuf = IoBuffer.allocate(buf.capacity()).setAutoExpand(true);
242 
243         remainingBuf.order(buf.order());
244         remainingBuf.put(buf);
245 
246         session.setAttribute(BUFFER, remainingBuf);
247     }
248     
249     /**
250      * Let the user change the way we handle fragmentation. If set to <tt>false</tt>, the 
251      * decode() method will not check the TransportMetadata fragmentation capability
252      *  
253      * @param transportMetadataFragmentation The flag to set.
254      */
255     public void setTransportMetadataFragmentation(boolean transportMetadataFragmentation) {
256         this.transportMetadataFragmentation = transportMetadataFragmentation;
257     }
258 }