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.handler.support;
    21
    22  import java.io.IOException;
    23  import java.io.InputStream;
    24
    25  import org.apache.mina.common.ByteBuffer;
    26  import org.apache.mina.common.IoHandler;
    27  import org.apache.mina.common.IoSession;
    28
    29  /**
    30   * An {@link InputStream} that buffers data read from
    31   * {@link IoHandler#messageReceived(IoSession, Object)} events.
    32   *
    33   * @author The Apache Directory Project (mina-dev@directory.apache.org)
    34   * @version $Rev: 555855 $, $Date: 2007-07-13 05:19:00 +0200 (Fri, 13 Jul 2007) $
    35   */
    36  public class IoSessionInputStream extends InputStream {
    37      private final Object mutex = new Object();
    38
    39      private final ByteBuffer buf;
    40
    41      private volatile boolean closed;
    42
    43      private volatile boolean released;
    44
    45      private IOException exception;
    46
    47      public IoSessionInputStream() {
    48          buf = ByteBuffer.allocate(16);
    49          buf.setAutoExpand(true);
    50          buf.limit(0);
    51      }
    52
    53      public int available() {
    54          if (released) {
    55              return 0;
    56          } else {
    57              synchronized (mutex) {
    58                  return buf.remaining();
    59              }
    60          }
    61      }
    62
    63      public void close() {
    64          if (closed) {
    65              return;
    66          }
    67
    68          synchronized (mutex) {
    69              closed = true;
    70              releaseBuffer();
    71
    72              mutex.notifyAll();
    73          }
    74      }
    75
    76      public int read() throws IOException {
    77          synchronized (mutex) {
    78              if (!waitForData()) {
    79                  return -1;
    80              }
    81
    82              return buf.get() & 0xff;
    83          }
    84      }
    85
    86      public int read(byte[] b, int off, int len) throws IOException {
    87          synchronized (mutex) {
    88              if (!waitForData()) {
    89                  return -1;
    90              }
    91
    92              int readBytes;
    93
    94              if (len > buf.remaining()) {
    95                  readBytes = buf.remaining();
    96              } else {
    97                  readBytes = len;
    98              }
    99
    100             buf.get(b, off, readBytes);
    101
    102             return readBytes;
    103         }
    104     }
    105
    106     private boolean waitForData() throws IOException {
    107         if (released) {
    108             return false;
    109         }
    110
    111         synchronized (mutex) {
    112             while (!released && buf.remaining() == 0 && exception == null) {
    113                 try {
    114                     mutex.wait();
    115                 } catch (InterruptedException e) {
    116                     IOException ioe = new IOException(
    117                             "Interrupted while waiting for more data");
    118                     ioe.initCause(e);
    119                     throw ioe;
    120                 }
    121             }
    122         }
    123
    124         if (exception != null) {
    125             releaseBuffer();
    126             throw exception;
    127         }
    128
    129         if (closed && buf.remaining() == 0) {
    130             releaseBuffer();
    131
    132             return false;
    133         }
    134
    135         return true;
    136     }
    137
    138     private void releaseBuffer() {
    139         if (released) {
    140             return;
    141         }
    142
    143         released = true;
    144         buf.release();
    145     }
    146
    147     public void write(ByteBuffer src) {
    148         synchronized (mutex) {
    149             if (closed) {
    150                 return;
    151             }
    152
    153             if (buf.hasRemaining()) {
    154                 this.buf.compact();
    155                 this.buf.put(src);
    156                 this.buf.flip();
    157             } else {
    158                 this.buf.clear();
    159                 this.buf.put(src);
    160                 this.buf.flip();
    161                 mutex.notifyAll();
    162             }
    163         }
    164     }
    165
    166     public void throwException(IOException e) {
    167         synchronized (mutex) {
    168             if (exception == null) {
    169                 exception = e;
    170
    171                 mutex.notifyAll();
    172             }
    173         }
    174     }
    175 }