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' && 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 AttributeKeytributeKey.html#AttributeKey">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"../../../../../org/apache/mina/core/buffer/IoBuffer.html#IoBuffer">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 }