add PipelineStream

Author: ymh199478

src/CatLib.Core.Tests/CatLib.Core.Tests.csproj

@@ -52,6 +52,7 @@
     <Compile Include="Support\RingBuffer\RingBufferTests.cs" />
     <Compile Include="Support\SortSet\SortSetTests.cs" />
     <Compile Include="Support\Storage\MemoryStorageTests.cs" />
+    <Compile Include="Support\Stream\PipelineStreamTests.cs" />
     <Compile Include="Support\Stream\StorageStreamTests.cs" />
     <Compile Include="Support\Template\ManagerTests.cs" />
     <Compile Include="Support\Template\SingleManagerTests.cs" />

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

@@ -0,0 +1,152 @@
+/*
+ * This file is part of the CatLib package.
+ *
+ * (c) Yu Bin <[email protected]>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ *
+ * Document: http://catlib.io/
+ */
+
+using System;
+using System.IO;
+using System.Text;
+using System.Threading;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+namespace CatLib.Core.Tests.Support.Stream
+{
+    [TestClass]
+    public class PipelineStreamTests
+    {
+        private PipelineStream stream;
+
+        [TestMethod]
+        public void TestPipeline()
+        {
+            stream = new PipelineStream(256);
+            ThreadPool.QueueUserWorkItem(WriteThread);
+
+            var wrote = false;
+            stream.OnWrote += (_) =>
+            {
+                wrote = true;
+            };
+            var data = new byte[100];
+            int read;
+            var actual = new StringBuilder();
+            var rand = new Random();
+            while ((read = stream.Read(data, 0, data.Length)) != 0)
+            {
+                var str = Encoding.UTF8.GetString(data, 0, read);
+                actual.Append(str);
+                Thread.Sleep(rand.Next(1, 5));
+            }
+            stream.Dispose();
+
+            var expected = new StringBuilder();
+            for (var i = 0; i < 1000; i++)
+            {
+                expected.Append("0123456789");
+            }
+           
+            Assert.AreEqual(expected.ToString(), actual.ToString());
+            Assert.AreEqual(true, wrote);
+        }
+
+        public void WriteThread(object obj)
+        {
+            var data = Encoding.UTF8.GetBytes("0123456789");
+            for (var i = 0; i < 1000; i++)
+            {
+                stream.Write(data, 0, data.Length);
+            }
+            stream.Close();
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(ObjectDisposedException))]
+        public void TestClosedAndWrite()
+        {
+            var stream = new PipelineStream(256);
+            Assert.AreEqual(true, stream.CanWrite);
+            stream.Write(Encoding.UTF8.GetBytes("0123456789"), 0, 10);
+            stream.Close();
+            Assert.AreEqual(false, stream.CanWrite);
+            Assert.AreEqual(true, stream.CanRead);
+            Assert.AreEqual(true, stream.Closed);
+            stream.Write(Encoding.UTF8.GetBytes("0123456789"), 0, 10);
+        }
+
+
+        [TestMethod]
+        [ExpectedException(typeof(ObjectDisposedException))]
+        public void TestDisposeAndWrite()
+        {
+            var stream = new PipelineStream(256);
+            Assert.AreEqual(true, stream.CanWrite);
+            Assert.AreEqual(false, stream.CanRead);
+            stream.Dispose();
+            Assert.AreEqual(false, stream.CanWrite);
+            Assert.AreEqual(false, stream.CanRead);
+            stream.Write(Encoding.UTF8.GetBytes("0123456789"), 0, 10);
+        }
+
+        [TestMethod]
+        public void TestCanReadCanWrite()
+        {
+            var stream = new PipelineStream(256);
+            stream.Write(Encoding.UTF8.GetBytes("0123456789"), 0, 10);
+            Assert.AreEqual(true, stream.CanWrite);
+            Assert.AreEqual(true, stream.CanRead);
+        }
+
+        [TestMethod]
+        public void TestCanSeek()
+        {
+            var stream = new PipelineStream(256);
+            Assert.AreEqual(false, stream.CanSeek);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(NotSupportedException))]
+        public void TestSeek()
+        {
+            var stream = new PipelineStream(256);
+            stream.Seek(0, SeekOrigin.Begin);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(NotSupportedException))]
+        public void TestSetLength()
+        {
+            var stream = new PipelineStream(256);
+            stream.SetLength(0);
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(NotSupportedException))]
+        public void TestSetPosition()
+        {
+            var stream = new PipelineStream(256);
+            stream.Position = 10;
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(NotSupportedException))]
+        public void TestGetPosition()
+        {
+            var stream = new PipelineStream(256);
+            var pos = stream.Position;
+        }
+
+        [TestMethod]
+        [ExpectedException(typeof(NotSupportedException))]
+        public void TestGetLength()
+        {
+            var stream = new PipelineStream(256);
+            var length = stream.Length;
+        }
+    }
+}

src/CatLib.Core/CatLib.Core.csproj

@@ -86,6 +86,7 @@
     <Compile Include="Support\RingBuffer\IRingBuffer.cs" />
     <Compile Include="Support\RingBuffer\RingBuffer.cs" />
     <Compile Include="Support\Storage\IStorage.cs" />
+    <Compile Include="Support\Stream\PipelineStream.cs" />
     <Compile Include="Support\Stream\StorageStream.cs" />
     <Compile Include="Support\Storage\MemoryStorage.cs" />
     <Compile Include="Support\Util\Dict.cs" />