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.common.ByteBuffer;
    23  import org.apache.mina.common.ByteBufferProxy;
    24  import org.apache.mina.common.IoFilter;
    25  import org.apache.mina.common.IoFilterAdapter;
    26  import org.apache.mina.common.IoFilterChain;
    27  import org.apache.mina.common.IoSession;
    28  import org.apache.mina.common.WriteFuture;
    29  import org.apache.mina.common.support.DefaultWriteFuture;
    30  import org.apache.mina.filter.codec.support.SimpleProtocolDecoderOutput;
    31  import org.apache.mina.filter.codec.support.SimpleProtocolEncoderOutput;
    32  import org.apache.mina.util.SessionLog;
    33
    34  /**
    35   * An {@link IoFilter} which translates binary or protocol specific data into
    36   * message object and vice versa using {@link ProtocolCodecFactory},
    37   * {@link ProtocolEncoder}, or {@link ProtocolDecoder}.
    38   *
    39   * @author The Apache Directory Project (mina-dev@directory.apache.org)
    40   * @version $Rev: 612026 $, $Date: 2008-01-15 07:16:14 +0100 (Tue, 15 Jan 2008) $
    41   */
    42  public class ProtocolCodecFilter extends IoFilterAdapter {
    43      public static final String ENCODER = ProtocolCodecFilter.class.getName()
    44              + ".encoder";
    45
    46      public static final String DECODER = ProtocolCodecFilter.class.getName()
    47              + ".decoder";
    48
    49      private static final String DECODER_OUT = ProtocolCodecFilter.class.getName()
    50              + ".decoderOut";
    51
    52      private static final Class<?>[] EMPTY_PARAMS = new Class[0];
    53
    54      private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.wrap(new byte[0]);
    55
    56      private final ProtocolCodecFactory factory;
    57
    58      public ProtocolCodecFilter(ProtocolCodecFactory factory) {
    59          if (factory == null) {
    60              throw new NullPointerException("factory");
    61          }
    62          this.factory = factory;
    63      }
    64
    65      public ProtocolCodecFilter(final ProtocolEncoder encoder,
    66              final ProtocolDecoder decoder) {
    67          if (encoder == null) {
    68              throw new NullPointerException("encoder");
    69          }
    70          if (decoder == null) {
    71              throw new NullPointerException("decoder");
    72          }
    73
    74          this.factory = new ProtocolCodecFactory() {
    75              public ProtocolEncoder getEncoder() {
    76                  return encoder;
    77              }
    78
    79              public ProtocolDecoder getDecoder() {
    80                  return decoder;
    81              }
    82          };
    83      }
    84
    85      public ProtocolCodecFilter(
    86              final Class<? extends ProtocolEncoder> encoderClass,
    87              final Class<? extends ProtocolDecoder> decoderClass) {
    88          if (encoderClass == null) {
    89              throw new NullPointerException("encoderClass");
    90          }
    91          if (decoderClass == null) {
    92              throw new NullPointerException("decoderClass");
    93          }
    94          if (!ProtocolEncoder.class.isAssignableFrom(encoderClass)) {
    95              throw new IllegalArgumentException("encoderClass: "
    96                      + encoderClass.getName());
    97          }
    98          if (!ProtocolDecoder.class.isAssignableFrom(decoderClass)) {
    99              throw new IllegalArgumentException("decoderClass: "
    100                     + decoderClass.getName());
    101         }
    102         try {
    103             encoderClass.getConstructor(EMPTY_PARAMS);
    104         } catch (NoSuchMethodException e) {
    105             throw new IllegalArgumentException(
    106                     "encoderClass doesn't have a public default constructor.");
    107         }
    108         try {
    109             decoderClass.getConstructor(EMPTY_PARAMS);
    110         } catch (NoSuchMethodException e) {
    111             throw new IllegalArgumentException(
    112                     "decoderClass doesn't have a public default constructor.");
    113         }
    114
    115         this.factory = new ProtocolCodecFactory() {
    116             public ProtocolEncoder getEncoder() throws Exception {
    117                 return encoderClass.newInstance();
    118             }
    119
    120             public ProtocolDecoder getDecoder() throws Exception {
    121                 return decoderClass.newInstance();
    122             }
    123         };
    124     }
    125
    126     @Override
    127     public void onPreAdd(IoFilterChain parent, String name,
    128             NextFilter nextFilter) throws Exception {
    129         if (parent.contains(ProtocolCodecFilter.class)) {
    130             throw new IllegalStateException(
    131                     "A filter chain cannot contain more than one ProtocolCodecFilter.");
    132         }
    133     }
    134
    135     @Override
    136     public void onPostRemove(IoFilterChain parent, String name,
    137             NextFilter nextFilter) throws Exception {
    138         disposeEncoder(parent.getSession());
    139         disposeDecoder(parent.getSession());
    140         disposeDecoderOut(parent.getSession());
    141     }
    142
    143     @Override
    144     public void messageReceived(NextFilter nextFilter, IoSession session,
    145             Object message) throws Exception {
    146         if (!(message instanceof ByteBuffer)) {
    147             nextFilter.messageReceived(session, message);
    148             return;
    149         }
    150
    151         ByteBuffer in = (ByteBuffer) message;
    152         ProtocolDecoder decoder = getDecoder(session);
    153         ProtocolDecoderOutput decoderOut = getDecoderOut(session, nextFilter);
    154
    155         int oldPos = in.position();
    156         try {
    157             synchronized (decoderOut) {
    158                 decoder.decode(session, in, decoderOut);
    159             }
    160         } catch (Throwable t) {
    161             ProtocolDecoderException pde;
    162             if (t instanceof ProtocolDecoderException) {
    163                 pde = (ProtocolDecoderException) t;
    164             } else {
    165                 pde = new ProtocolDecoderException(t);
    166             }
    167
    168             if (pde.getHexdump() == null) {
    169                 int curPos = in.position();
    170                 in.position(oldPos);
    171                 pde.setHexdump(in.getHexDump());
    172                 in.position(curPos);
    173             }
    174             throw pde;
    175         } finally {
    176             try {
    177                 // Release the read buffer.
    178                 in.release();
    179             } finally {
    180                 decoderOut.flush();
    181             }
    182         }
    183     }
    184
    185     @Override
    186     public void messageSent(NextFilter nextFilter, IoSession session,
    187             Object message) throws Exception {
    188         if (message instanceof HiddenByteBuffer) {
    189             return;
    190         }
    191
    192         if (!(message instanceof MessageByteBuffer)) {
    193             nextFilter.messageSent(session, message);
    194             return;
    195         }
    196
    197         nextFilter.messageSent(session, ((MessageByteBuffer) message).message);
    198     }
    199
    200     @Override
    201     public void filterWrite(NextFilter nextFilter, IoSession session,
    202             WriteRequest writeRequest) throws Exception {
    203         Object message = writeRequest.getMessage();
    204         if (message instanceof ByteBuffer) {
    205             nextFilter.filterWrite(session, writeRequest);
    206             return;
    207         }
    208
    209         ProtocolEncoder encoder = getEncoder(session);
    210         ProtocolEncoderOutputImpl encoderOut = getEncoderOut(session,
    211                 nextFilter, writeRequest);
    212
    213         try {
    214             encoder.encode(session, message, encoderOut);
    215             encoderOut.flush();
    216             nextFilter.filterWrite(session, new WriteRequest(
    217                     new MessageByteBuffer(writeRequest.getMessage()),
    218                     writeRequest.getFuture(), writeRequest.getDestination()));
    219         } catch (Throwable t) {
    220             ProtocolEncoderException pee;
    221             if (t instanceof ProtocolEncoderException) {
    222                 pee = (ProtocolEncoderException) t;
    223             } else {
    224                 pee = new ProtocolEncoderException(t);
    225             }
    226             throw pee;
    227         }
    228     }
    229
    230     @Override
    231     public void sessionClosed(NextFilter nextFilter, IoSession session)
    232             throws Exception {
    233         // Call finishDecode() first when a connection is closed.
    234         ProtocolDecoder decoder = getDecoder(session);
    235         ProtocolDecoderOutput decoderOut = getDecoderOut(session, nextFilter);
    236         try {
    237             decoder.finishDecode(session, decoderOut);
    238         } catch (Throwable t) {
    239             ProtocolDecoderException pde;
    240             if (t instanceof ProtocolDecoderException) {
    241                 pde = (ProtocolDecoderException) t;
    242             } else {
    243                 pde = new ProtocolDecoderException(t);
    244             }
    245             throw pde;
    246         } finally {
    247             // Dispose all.
    248             disposeEncoder(session);
    249             disposeDecoder(session);
    250             disposeDecoderOut(session);
    251             decoderOut.flush();
    252         }
    253
    254         nextFilter.sessionClosed(session);
    255     }
    256
    257     private ProtocolEncoder getEncoder(IoSession session) throws Exception {
    258         ProtocolEncoder encoder = (ProtocolEncoder) session
    259                 .getAttribute(ENCODER);
    260         if (encoder == null) {
    261             encoder = factory.getEncoder();
    262             session.setAttribute(ENCODER, encoder);
    263         }
    264         return encoder;
    265     }
    266
    267     private ProtocolEncoderOutputImpl getEncoderOut(IoSession session,
    268             NextFilter nextFilter, WriteRequest writeRequest) {
    269         return new ProtocolEncoderOutputImpl(session, nextFilter, writeRequest);
    270     }
    271
    272     private ProtocolDecoder getDecoder(IoSession session) throws Exception {
    273         ProtocolDecoder decoder = (ProtocolDecoder) session
    274                 .getAttribute(DECODER);
    275         if (decoder == null) {
    276             decoder = factory.getDecoder();
    277             session.setAttribute(DECODER, decoder);
    278         }
    279         return decoder;
    280     }
    281
    282     private ProtocolDecoderOutput getDecoderOut(IoSession session,
    283             NextFilter nextFilter) {
    284         ProtocolDecoderOutput out = (ProtocolDecoderOutput) session.getAttribute(DECODER_OUT);
    285         if (out == null) {
    286             out = new SimpleProtocolDecoderOutput(session, nextFilter);
    287             session.setAttribute(DECODER_OUT, out);
    288         }
    289         return out;
    290     }
    291
    292     private void disposeEncoder(IoSession session) {
    293         ProtocolEncoder encoder = (ProtocolEncoder) session
    294                 .removeAttribute(ENCODER);
    295         if (encoder == null) {
    296             return;
    297         }
    298
    299         try {
    300             encoder.dispose(session);
    301         } catch (Throwable t) {
    302             SessionLog.warn(session, "Failed to dispose: "
    303                     + encoder.getClass().getName() + " (" + encoder + ')');
    304         }
    305     }
    306
    307     private void disposeDecoder(IoSession session) {
    308         ProtocolDecoder decoder = (ProtocolDecoder) session
    309                 .removeAttribute(DECODER);
    310         if (decoder == null) {
    311             return;
    312         }
    313
    314         try {
    315             decoder.dispose(session);
    316         } catch (Throwable t) {
    317             SessionLog.warn(session, "Falied to dispose: "
    318                     + decoder.getClass().getName() + " (" + decoder + ')');
    319         }
    320     }
    321
    322     private void disposeDecoderOut(IoSession session) {
    323         session.removeAttribute(DECODER_OUT);
    324     }
    325
    326     private static class HiddenByteBuffer extends ByteBufferProxy {
    327         private HiddenByteBuffer(ByteBuffer buf) {
    328             super(buf);
    329         }
    330     }
    331
    332     private static class MessageByteBuffer extends ByteBufferProxy {
    333         private final Object message;
    334
    335         private MessageByteBuffer(Object message) {
    336             super(EMPTY_BUFFER);
    337             this.message = message;
    338         }
    339
    340         @Override
    341         public void acquire() {
    342             // no-op since we are wraping a zero-byte buffer, this instance is to just curry the message
    343         }
    344
    345         @Override
    346         public void release() {
    347             // no-op since we are wraping a zero-byte buffer, this instance is to just curry the message
    348         }
    349     }
    350
    351     private static class ProtocolEncoderOutputImpl extends
    352             SimpleProtocolEncoderOutput {
    353         private final IoSession session;
    354
    355         private final NextFilter nextFilter;
    356
    357         private final WriteRequest writeRequest;
    358
    359         ProtocolEncoderOutputImpl(IoSession session, NextFilter nextFilter,
    360                 WriteRequest writeRequest) {
    361             this.session = session;
    362             this.nextFilter = nextFilter;
    363             this.writeRequest = writeRequest;
    364         }
    365
    366         @Override
    367         protected WriteFuture doFlush(ByteBuffer buf) {
    368             WriteFuture future = new DefaultWriteFuture(session);
    369             nextFilter.filterWrite(session, new WriteRequest(
    370                     new HiddenByteBuffer(buf), future, writeRequest
    371                             .getDestination()));
    372             return future;
    373         }
    374     }
    375 }