supported set length

Author: ymh199478

src/CatLib.Core.Tests/Support/Stream/PipelineStreamTests.cs

@@ -28,10 +28,10 @@ public void TestPipeline()
             stream = new PipelineStream(256);
             ThreadPool.QueueUserWorkItem(WriteThread);
 
-            var wrote = false;
-            stream.OnWrote += (_) =>
+            var readed = false;
+            stream.OnRead += (_) =>
             {
-                wrote = true;
+                readed = true;
             };
             var data = new byte[100];
             int read;
@@ -52,7 +52,7 @@ public void TestPipeline()
             }
 
             Assert.AreEqual(expected.ToString(), actual.ToString());
-            Assert.AreEqual(true, wrote);
+            Assert.AreEqual(true, readed);
         }
 
         public void WriteThread(object obj)

src/CatLib.Core/Support/Stream/PipelineStream.cs

@@ -42,9 +42,9 @@ public class PipelineStream : Stream
         private readonly RingBuffer ringBuffer;
 
         /// <summary>
-        /// 当写入完成后触发
+        /// 当完成读取后触发
         /// </summary>
-        public event Action<Stream> OnWrote;
+        public event Action<Stream> OnRead;
 
         /// <summary>
         /// 是否已经被释放了
@@ -56,11 +56,6 @@ public class PipelineStream : Stream
         /// </summary>
         private volatile bool closed;
 
-        /// <summary>
-        /// 总流量
-        /// </summary>
-        public long TotalFlow { get; private set; }
-
         /// <summary>
         /// 是否可以被读取
         /// </summary>
@@ -77,21 +72,31 @@ public override bool CanWrite
             get { return count < capacity && !closed; }
         }
 
+        /// <summary>
+        /// 当前流的位置
+        /// </summary>
+        private long position;
+
         /// <summary>
         /// 偏移位置（不支持）
         /// </summary>
         public override long Position
         {
-            get { throw new NotSupportedException(); }
+            get { return position; }
             set { throw new NotSupportedException(); }
         }
 
+        /// <summary>
+        /// 流的长度
+        /// </summary>
+        private long length;
+
         /// <summary>
         /// 流的长度
         /// </summary>
         public override long Length
         {
-            get { throw new NotSupportedException(); }
+            get { return length; }
         }
 
         /// <summary>
@@ -119,7 +124,6 @@ public PipelineStream(int capacity = 4096, int sleep = 1)
         {
             this.capacity = capacity.ToPrime();
             this.sleep = Math.Max(0, sleep);
-            TotalFlow = 0;
             ringBuffer = new RingBuffer(this.capacity, false);
         }
 
@@ -148,7 +152,7 @@ public override long Seek(long offset, SeekOrigin origin)
         /// <param name="value">长度</param>
         public override void SetLength(long value)
         {
-            throw new NotSupportedException();
+            length = Math.Max(0, value);
         }
 
         /// <summary>
@@ -198,7 +202,13 @@ public override int Read(byte[] buffer, int offset, int count)
                     {
                         var read = ringBuffer.Read(buffer, offset, count);
                         this.count -= read;
-                        TotalFlow += read;
+                        position += read;
+
+                        if (OnRead != null)
+                        {
+                            OnRead(this);
+                        }
+
                         return read;
                     }
                     finally
@@ -242,12 +252,6 @@ public override void Write(byte[] buffer, int offset, int count)
 
                     Guard.Requires<AssertException>(ringBuffer.Write(buffer, offset, count) == count);
                     this.count += count;
-
-                    if (OnWrote != null)
-                    {
-                        OnWrote(this);
-                    }
-
                     return;
                 }
             }