LengthMarkedBufferedInputStream.cs

Go to the documentation of this file.

using System;
using System.IO;

using Deveel.Data.Util;

namespace Deveel.Data.Protocol {
    public sealed class LengthMarkedBufferedInputStream : Stream {
        private const int InitialBufferSize = 512;

        private readonly IInputStream input;

        private byte[] buf;

        private int count;

        private int markedLength;

        private int markedIndex;

        public LengthMarkedBufferedInputStream(IInputStream input) {
            this.input = input;
            buf = new byte[InitialBufferSize];
            count = 0;
            markedLength = -1;
            markedIndex = -1;
        }

        private void EnsureCapacity(int newSize) {
            int old_size = buf.Length;
            if (newSize > old_size) {
                int cap = (old_size * 3) / 2 + 1;
                if (cap < newSize)
                    cap = newSize;
                byte[] oldBuf = buf;
                buf = new byte[cap];
                //      // Copy all the contents except the first 4 bytes (the size marker)
                Array.Copy(oldBuf, 0, buf, 0, count);
            }
        }

        private void HandleEndReached() {
            // Move anything from the end of the buffer to the start.
            Array.Copy(buf, markedIndex, buf, 0, count - markedLength);
            count -= markedLength;

            // Reset the state
            markedLength = -1;
            markedIndex = -1;
        }

        // ---------- Overwritten from Stream ----------

        public override int ReadByte() {
            lock (this) {
                if (markedIndex == -1)
                    throw new IOException("No mark has been read yet.");

                if (markedIndex >= markedLength) {
                    string debugMsg = "Read over end of length marked buffer.  ";
                    debugMsg += "(marked_index=" + markedIndex;
                    debugMsg += ",marked_length=" + markedLength + ")";
                    debugMsg += ")";
                    throw new IOException(debugMsg);
                }
                int n = buf[markedIndex++] & 0x0FF;
                if (markedIndex >= markedLength) {
                    HandleEndReached();
                }
                return n;
            }
        }

        public override void Write(byte[] buffer, int offset, int count) {
            throw new NotSupportedException();
        }

        public override bool CanRead {
            get { return true; }
        }

        public override bool CanSeek {
            get { return false; }
        }

        public override bool CanWrite {
            get { return false; }
        }

        public override long Length {
            get { throw new NotImplementedException(); }
        }

        public override long Position {
            get { throw new NotImplementedException(); }
            set { throw new NotImplementedException(); }
        }

        public override void Flush() {
        }

        public override long Seek(long offset, SeekOrigin origin) {
            throw new NotSupportedException();
        }

        public override void SetLength(long value) {
            throw new NotSupportedException();
        }

        public override int Read(byte[] b, int off, int len) {
            lock (this) {
                if (markedIndex == -1)
                    throw new IOException("No mark has been read yet.");

                int readUpto = markedIndex + len;
                if (readUpto > markedLength) {
                    String debug_msg = "Read over end of length marked buffer.  ";
                    debug_msg += "(marked_index=" + markedIndex;
                    debug_msg += ",len=" + len;
                    debug_msg += ",marked_length=" + markedLength + ")";
                    throw new IOException(debug_msg);
                }
                Array.Copy(buf, markedIndex, b, off, len);
                markedIndex = readUpto;
                if (markedIndex >= markedLength) {
                    HandleEndReached();
                }
                return len;
            }
        }

        public int Available {
            get {
                lock (this) {
                    // This method only returns a non 0 value if there is a complete command
                    // waiting on the stream.
                    if (markedLength >= 0) {
                        return (markedLength - markedIndex);
                    }
                    return 0;
                }
            }
        }

        // ---------- These methods aid in reading state from the stream ----------

        public bool PollForCommand(int maxSize) {
            lock (this) {
                if (markedLength == -1) {
                    int available = input.Available;
                    if (count > 0 || available > 0) {
                        if ((count + available) > maxSize) {
                            throw new IOException("Marked length is greater than max size ( " +
                                                  (count + available) + " > " + maxSize + " )");
                        }

                        EnsureCapacity(count + available);
                        int readIn = input.Read(buf, count, available);

                        if (readIn == 0) {
                            //TODO: Check this format...
                            // throw new EndOfStreamException();

                            // zero bytes read means that the stream is finished...
                            return false;
                        }
                        count = count + readIn;

                        // Check: Is a complete command available?
                        if (count >= 4) {
                            int lengthMarker = ByteBuffer.ReadInt4(buf, 0);

                            if (count >= lengthMarker + 4) {
                                // Yes, complete command available.
                                // mark this area up.
                                markedLength = lengthMarker + 4;
                                markedIndex = 4;
                                return true;
                            }
                        }
                    }
                }
                return false;
            }
        }

        public void BlockForCommand() {
            lock (this) {
                while (true) {
                    // Is there a command available?
                    if (count >= 4) {
                        int lengthMarker = ByteBuffer.ReadInt4(buf, 0);
                        if (count >= lengthMarker + 4) {
                            // Yes, complete command available.
                            // mark this area up.
                            markedLength = lengthMarker + 4;
                            markedIndex = 4;
                            return;
                        }
                    }

                    // If the buffer is full grow it larger.
                    if (count >= buf.Length) {
                        EnsureCapacity(count + InitialBufferSize);
                    }
                    // Read in a block of data, block if nothing there
                    int read_in = input.Read(buf, count, buf.Length - count);
                    if (read_in == 0) {
                        //TODO: Check this format...
                        // throw new EndOfStreamException();

                        // zero bytes read means that the stream is finished...
                        return;
                    }
                    count += read_in;
                }
            }
        }
    }
}