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
29   * buffers to a <em>cumulative buffer</em> to help users implement decoders.
30   * <p>
31   * If the received {@link IoBuffer} is only a part of a message.
32   * decoders should cumulate received buffers to make a message complete or
33   * to postpone decoding until more buffers arrive.
34   * <p>
35   * Here is an example decoder that decodes CRLF terminated lines into
36   * <code>Command</code> objects:
37   * <pre>
38   * public class CrLfTerminatedCommandLineDecoder
39   *         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(
48   *             IoSession session, IoBuffer in, ProtocolDecoderOutput out)
49   *             throws Exception {
50   *
51   *         // Remember the initial position.
52   *         int start = in.position();
53   *
54   *         // Now find the first CRLF in the buffer.
55   *         byte previous = 0;
56   *         while (in.hasRemaining()) {
57   *             byte current = in.get();
58   *
59   *             if (previous == '\r' && current == '\n') {
60   *                 // Remember the current position and limit.
61   *                 int position = in.position();
62   *                 int limit = in.limit();
63   *                 try {
64   *                     in.position(start);
65   *                     in.limit(position);
66   *                     // The bytes between in.position() and in.limit()
67   *                     // now contain a full CRLF terminated line.
68   *                     out.write(parseCommand(in.slice()));
69   *                 } finally {
70   *                     // Set the position to point right after the
71   *                     // detected line and set the limit to the old
72   *                     // one.
73   *                     in.position(position);
74   *                     in.limit(limit);
75   *                 }
76   *                 // Decoded one line; CumulativeProtocolDecoder will
77   *                 // call me again until I return false. So just
78   *                 // return true until there are no more lines in the
79   *                 // buffer.
80   *                 return true;
81   *             }
82   *
83   *             previous = current;
84   *         }
85   *
86   *         // Could not find CRLF in the buffer. Reset the initial
87   *         // position to the one we recorded above.
88   *         in.position(start);
89   *
90   *         return false;
91   *     }
92   * }
93   * </pre>
94   * <p>
95   * Please note that this decoder simply forward the call to
96   * {@link #doDecode(IoSession, IoBuffer, ProtocolDecoderOutput)} if the
97   * underlying transport doesn't have a packet fragmentation.  Whether the
98   * transport has fragmentation or not is determined by querying
99   * {@link TransportMetadata}.
100  *
101  * @author <a href="http://mina.apache.org">Apache MINA Project</a>
102  */
103 public abstract class CumulativeProtocolDecoder extends ProtocolDecoderAdapter {
104 
105     private final AttributeKey BUFFER = new AttributeKey(getClass(), "buffer");
106 
107     /**
108      * Creates a new instance.
109      */
110     protected CumulativeProtocolDecoder() {
111         // Do nothing
112     }
113 
114     /**
115      * Cumulates content of <tt>in</tt> into internal buffer and forwards
116      * decoding request to {@link #doDecode(IoSession, IoBuffer, ProtocolDecoderOutput)}.
117      * <tt>doDecode()</tt> is invoked repeatedly until it returns <tt>false</tt>
118      * and the cumulative buffer is compacted after decoding ends.
119      *
120      * @throws IllegalStateException if your <tt>doDecode()</tt> returned
121      *                               <tt>true</tt> not consuming the cumulative buffer.
122      */
123     public void decode(IoSession session, IoBuffer in,
124             ProtocolDecoderOutput out) throws Exception {
125         if (!session.getTransportMetadata().hasFragmentation()) {
126             while (in.hasRemaining()) {
127                 if (!doDecode(session, in, out)) {
128                     break;
129                 }
130             }
131 
132             return;
133         }
134 
135         boolean usingSessionBuffer = true;
136         IoBuffer buf = (IoBuffer) session.getAttribute(BUFFER);
137         // If we have a session buffer, append data to that; otherwise
138         // use the buffer read from the network directly.
139         if (buf != null) {
140             boolean appended = false;
141             // Make sure that the buffer is auto-expanded.
142             if (buf.isAutoExpand()) {
143                 try {
144                     buf.put(in);
145                     appended = true;
146                 } catch (IllegalStateException e) {
147                     // A user called derivation method (e.g. slice()),
148                     // which disables auto-expansion of the parent buffer.
149                 } catch (IndexOutOfBoundsException e) {
150                     // A user disabled auto-expansion.
151                 }
152             }
153 
154             if (appended) {
155                 buf.flip();
156             } else {
157                 // Reallocate the buffer if append operation failed due to
158                 // derivation or disabled auto-expansion.
159                 buf.flip();
160                 IoBuffer newBuf = IoBuffer.allocate(
161                         buf.remaining() + in.remaining()).setAutoExpand(true);
162                 newBuf.order(buf.order());
163                 newBuf.put(buf);
164                 newBuf.put(in);
165                 newBuf.flip();
166                 buf = newBuf;
167 
168                 // Update the session attribute.
169                 session.setAttribute(BUFFER, buf);
170             }
171         } else {
172             buf = in;
173             usingSessionBuffer = false;
174         }
175 
176         for (;;) {
177             int oldPos = buf.position();
178             boolean decoded = doDecode(session, buf, out);
179             if (decoded) {
180                 if (buf.position() == oldPos) {
181                     throw new IllegalStateException(
182                             "doDecode() can't return true when buffer is not consumed.");
183                 }
184 
185                 if (!buf.hasRemaining()) {
186                     break;
187                 }
188             } else {
189                 break;
190             }
191         }
192 
193         // if there is any data left that cannot be decoded, we store
194         // it in a buffer in the session and next time this decoder is
195         // invoked the session buffer gets appended to
196         if (buf.hasRemaining()) {
197             if (usingSessionBuffer && buf.isAutoExpand()) {
198                 buf.compact();
199             } else {
200                 storeRemainingInSession(buf, session);
201             }
202         } else {
203             if (usingSessionBuffer) {
204                 removeSessionBuffer(session);
205             }
206         }
207     }
208 
209     /**
210      * Implement this method to consume the specified cumulative buffer and
211      * decode its content into message(s).
212      *
213      * @param in the cumulative buffer
214      * @return <tt>true</tt> if and only if there's more to decode in the buffer
215      *         and you want to have <tt>doDecode</tt> method invoked again.
216      *         Return <tt>false</tt> if remaining data is not enough to decode,
217      *         then this method will be invoked again when more data is cumulated.
218      * @throws Exception if cannot decode <tt>in</tt>.
219      */
220     protected abstract boolean doDecode(IoSession session, IoBuffer in,
221             ProtocolDecoderOutput out) throws Exception;
222 
223     /**
224      * Releases the cumulative buffer used by the specified <tt>session</tt>.
225      * Please don't forget to call <tt>super.dispose( session )</tt> when
226      * you override this method.
227      */
228     @Override
229     public void dispose(IoSession session) throws Exception {
230         removeSessionBuffer(session);
231     }
232 
233     private void removeSessionBuffer(IoSession session) {
234         session.removeAttribute(BUFFER);
235     }
236 
237     private void storeRemainingInSession(IoBuffer buf, IoSession session) {
238         final IoBuffer remainingBuf = IoBuffer.allocate(buf.capacity()).setAutoExpand(true);
239 
240         remainingBuf.order(buf.order());
241         remainingBuf.put(buf);
242 
243         session.setAttribute(BUFFER, remainingBuf);
244     }
245 }