C:/work/OMMR/OCHHP_OMMR_OMMR/Code/Assemblies/SharpSSH/Source/SharpSSH-1.1.1.13.src/SharpSSH/Streams/PipedInputStream.cs

Go to the documentation of this file.

using System;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading;

namespace Tamir.Streams
{
/*
 * @(#)PipedInputStream.java    1.35 03/12/19
 *
 * Copyright 2004 Sun Microsystems, Inc. All rights reserved.
 * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */

        public class PipedInputStream : Tamir.SharpSsh.java.io.InputStream
        {
                internal bool closedByWriter = false;
                internal volatile bool closedByReader = false;
                internal bool connected = false;

                /* REMIND: identification of the read and write sides needs to be
                   more sophisticated.  Either using thread groups (but what about
                   pipes within a thread?) or using finalization (but it may be a
                   long time until the next GC). */
                internal Thread readSide;
                internal Thread writeSide;

                internal const int PIPE_SIZE = 1024;

                internal byte[] buffer = new byte[PIPE_SIZE];

                internal int m_in = -1;

                internal int m_out = 0;

                public PipedInputStream(PipedOutputStream src) 
                {
                        connect(src);
                }

                public PipedInputStream() 
                {
                        int i = 0;
                }

                public virtual void connect(PipedOutputStream src)
                {
                        src.connect(this);
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                internal void receive(int b) 
                {
                        checkStateForReceive();
                        writeSide = Thread.CurrentThread;
                        if (m_in == m_out)
                                awaitSpace();
                        if (m_in < 0) 
                        {
                                m_in = 0;
                                m_out = 0;
                        }
                        buffer[m_in++] = (byte)(b & 0xFF);
                        if (m_in >= buffer.Length) 
                        {
                                m_in = 0;
                        }
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                internal void receive(byte[] b, int off, int len) 
                {
                        checkStateForReceive();
                        writeSide = Thread.CurrentThread;
                        int bytesToTransfer = len;
                        while (bytesToTransfer > 0) 
                        {
                                if (m_in == m_out)
                                        awaitSpace();
                                int nextTransferAmount = 0;
                                if (m_out < m_in) 
                                {
                                        nextTransferAmount = buffer.Length - m_in;
                                } 
                                else if (m_in < m_out) 
                                {
                                        if (m_in == -1) 
                                        {
                                                m_in = m_out = 0;
                                                nextTransferAmount = buffer.Length - m_in;
                                        } 
                                        else 
                                        {
                                                nextTransferAmount = m_out - m_in;
                                        }
                                }
                                if (nextTransferAmount > bytesToTransfer)
                                        nextTransferAmount = bytesToTransfer;
                                assert(nextTransferAmount > 0);
                                Array.Copy(b, off, buffer, m_in, nextTransferAmount);
                                bytesToTransfer -= nextTransferAmount;
                                off += nextTransferAmount;
                                m_in += nextTransferAmount;
                                if (m_in >= buffer.Length) 
                                {
                                        m_in = 0;
                                }
                        }
                }

                private void checkStateForReceive()
                {
                        if (!connected) 
                        {
                                throw new IOException("Pipe not connected");
                        } 
                        else if (closedByWriter || closedByReader) 
                        {
                                throw new IOException("Pipe closed");
                        } 
                        else if (readSide != null && !readSide.IsAlive) 
                        {
                                throw new IOException("Read end dead");
                        }
                }

                private void awaitSpace()  
                {
                        while (m_in == m_out) 
                        {
                                if ((readSide != null) && !readSide.IsAlive) 
                                {
                                        throw new IOException("Pipe broken");
                                }
                                /* full: kick any waiting readers */
                                //java: notifyAll();
                                Monitor.PulseAll(this);
                                try 
                                {
                                        //java: wait(1000);
                                        Monitor.Wait(this, 1000);
                                } 
                                catch (ThreadInterruptedException  ex) 
                                {
                                        throw ex;
                                }
                        }
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                internal void receivedLast() 
                {
                        closedByWriter = true;
                        //notifyAll();
                        Monitor.PulseAll(this);
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                public virtual  int read()  
                {
                        if (!connected) 
                        {
                                throw new IOException("Pipe not connected");
                        } 
                        else if (closedByReader) 
                        {
                                throw new IOException("Pipe closed");
                        } 
                        else if (writeSide != null && !writeSide.IsAlive
                                && !closedByWriter && (m_in < 0)) 
                        {
                                throw new IOException("Write end dead");
                        }

                        readSide = Thread.CurrentThread;
                        int trials = 2;
                        while (m_in < 0) 
                        {
                                if (closedByWriter) 
                                {
                                        /* closed by writer, return EOF */
                                        return -1;
                                }
                                if ((writeSide != null) && (!writeSide.IsAlive) && (--trials < 0)) 
                                {
                                        throw new IOException("Pipe broken");
                                }
                                /* might be a writer waiting */
                                Monitor.PulseAll(this);
                                try 
                                {
                                        Monitor.Wait(this, 1000);
                                } 
                                catch (ThreadInterruptedException ex) 
                                {
                                        throw ex;
                                }
                        }
                        int ret = buffer[m_out++] & 0xFF;
                        if (m_out >= buffer.Length) 
                        {
                                m_out = 0;
                        }
                        if (m_in == m_out) 
                        {
                                /* now empty */
                                m_in = -1;
                        }
                        return ret;
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                public override int read(byte[] b, int off, int len)
                {
                        if (b == null) 
                        {
                                throw new NullReferenceException();
                        } 
                        else if ((off < 0) || (off > b.Length) || (len < 0) ||
                                ((off + len) > b.Length) || ((off + len) < 0)) 
                        {
                                throw new IndexOutOfRangeException();
                        } 
                        else if (len == 0) 
                        {
                                return 0;
                        }

                        /* possibly wait on the first character */
                        int c = read();
                        if (c < 0) 
                        {
                                return -1;
                        }
                        b[off] = (byte) c;
                        int rlen = 1;
                        while ((m_in >= 0) && (--len > 0)) 
                        {
                                b[off + rlen] = buffer[m_out++];
                                rlen++;
                                if (m_out >= buffer.Length) 
                                {
                                        m_out = 0;
                                }
                                if (m_in == m_out) 
                                {
                                        /* now empty */
                                        m_in = -1;
                                }
                        }
                        return rlen;
                }

                [MethodImpl(MethodImplOptions.Synchronized)]
                public virtual int available() 
                {
                        if(m_in < 0)
                                return 0;
                        else if(m_in == m_out)
                                return buffer.Length;
                        else if (m_in > m_out)
                                return m_in - m_out;
                        else
                                return m_in + buffer.Length - m_out;
                }

                public override void close()  
                {
                        closedByReader = true;
                        lock (this) 
                        {
                                m_in = -1;
                        }
                }

                private void assert(bool exp)
                {
                        if (!exp)
                                throw new Exception("Assertion failed!");
                }

                

                public override int Read(byte[] buffer, int offset, int count)
                {
                        return this.read(buffer, offset, count);
                }

                public override int ReadByte()
                {
                        return this.read();
                }
                
                public override void WriteByte(byte value)
                {
                }

                public override void Write(byte[] buffer, int offset, int count)
                {
                }
                public override void Close()
                {
                        base.Close ();
                        this.close();
                }
                public override bool CanRead
                {
                        get
                        {
                                return true;
                        }
                }
                public override bool CanWrite
                {
                        get
                        {
                                return false;
                        }
                }
                public override bool CanSeek
                {
                        get
                        {
                                return false;
                        }
                }
                public override void Flush()
                {
                        
                }
                public override long Length
                {
                        get
                        {
                                if(m_in > m_out)
                                        return (m_in - m_out);
                                else
                                {
                                        return (buffer.Length -m_out+m_in);
                                }
                        }
                }
                public override long Position
                {
                        get
                        {
                                return m_out;
                        }
                        set
                        {
                                throw new IOException("Setting the position of this stream is not supported");
                        }
                }
                public override void SetLength(long value)
                {
                        throw new IOException("Setting the length of this stream is not supported");
                }
                public override long Seek(long offset, SeekOrigin origin)
                {
                        return 0;
                }
        }
}

---