/*

  * #%L

  * JavaHg

  * %%

  * Copyright (C) 2011 aragost Trifork ag

  * %%

  * Permission is hereby granted, free of charge, to any person obtaining a copy

  * of this software and associated documentation files (the "Software"), to deal

  * in the Software without restriction, including without limitation the rights

  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

  * copies of the Software, and to permit persons to whom the Software is

  * furnished to do so, subject to the following conditions:

  * 

  * The above copyright notice and this permission notice shall be included in

  * all copies or substantial portions of the Software.

  * 

  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE

  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

  * THE SOFTWARE.

  * #L%

  */

 package com.aragost.javahg.internals;

 import java.io.IOException;

 import java.io.InputStream;

 import com.aragost.javahg.log.Logger;

 import com.aragost.javahg.log.LoggerFactory;

 /**

  * An input stream reading one channel block.

  */

 public class BlockInputStream extends InputStream {

     private static final Logger LOG = LoggerFactory.getLogger(BlockInputStream.class);

     public static final BlockInputStream EMPTY = new BlockInputStream();

     private final InputStream stream;

     /**

      * Name of channel

      */

     private final char channel;

     /**

      * Length of the block

      */

     private final int length;

     /**

      * Number of bytes left in the block

      */

     private int countDown;

     private BlockInputStream() {

         this.stream = null;

         this.channel = 'o';

         this.countDown = this.length = 0;

     }

     /**

      * Construct a new BlockInputStream. This will read the channel (1

      * byte) and length (4 bytes) from the argument InputStream.

      * 

      * @param stream

      * @throws IOException

      *             from reading channel and length from the underlying

      *             stream

      */

     BlockInputStream(InputStream stream) throws IOException {

         this.stream = stream;

         this.channel = (char) stream.read();

         this.length = this.countDown = readInt();

         if (this.countDown == -1) {

             throw new InvalidStreamException();

         }

         if (Character.isUpperCase(this.channel)) {

             // Uppercase means mandatory, which is assumed to be an

             // "input" channel

             this.countDown = 0;

         }

     }

     @Override

     public int read() throws IOException {

         if (countDown == 0) {

             return -1;

         } else {

             this.countDown--;

             int read = this.stream.read();

             LOG.debug("stdout char: {}", (char) read);

             return read;

         }

     }

     @Override

     public int read(byte[] b, int off, int len) throws IOException {

         if (this.countDown == 0) {

             return -1;

         }

         if (len > this.countDown) {

             len = this.countDown;

         }

         int result = stream.read(b, off, len);

         if (result == -1) {

             // The server should provide enough bytes. It told the

             // length up front

             // This can possible happen if the server is killed. Throw

             // an IOException so it is properly handled by the

             // Server#verifyServerProcess method

             throw new IOException("Unexpected EOF");

         }

         this.countDown -= result;

         if (LOG.isDebugEnabled()) {

             LOG.debug("read {}: '{}'", "" + this.channel + result, new String(b, off, result));

         }

         return result;

     }

     @Override

     public int available() throws IOException {

         if (this.countDown == 0) {

             return 0;

         }

         int underlyingAvailable = this.stream.available();

         return underlyingAvailable > this.countDown ? this.countDown : underlyingAvailable;

     }

     /**

      * @return int value (big endian) of the next 4 bytes from the

      *         underlying stream

      * @throws IOException

      */

     int readInt() throws IOException {

         return Utils.readBigEndian(this.stream);

     }

     char getChannel() {

         return this.channel;

     }

     int getLength() {

         return length;

     }

     int getBytesLeft() {

         return this.countDown;

     }

     /**

      * Exception to indicate that channel and length could not be read

      * from the underlying stream.

      */

     public static class InvalidStreamException extends RuntimeException {

         private static final long serialVersionUID = 1L;

     }

 }