feat(PubSubApi): publish a stream message

Author: richardschneider

src/CoreApi/PubSubApi.cs

@@ -56,7 +56,11 @@ internal PubSubApi(IpfsClient ipfs)
 
         public Task PublishAsync(string topic, Stream message, CancellationToken cancel = default(CancellationToken))
         {
-            throw new NotImplementedException();
+            using (MemoryStream ms = new MemoryStream())
+            {
+                message.CopyTo(ms);
+                return PublishAsync(topic, ms.ToArray(), cancel);
+            }
         }
 
         public async Task PublishAsync(string topic, string message, CancellationToken cancel = default(CancellationToken))

test/CoreApi/PubSubApiTest.cs

@@ -3,6 +3,7 @@
 using Newtonsoft.Json.Linq;
 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -200,5 +201,32 @@ await ipfs.PubSub.SubscribeAsync(topic, msg =>
                 cs.Cancel();
             }
         }
+
+        [TestMethod]
+        public async Task Subscribe_StreamMessage()
+        {
+            var messages = new List<IPublishedMessage>();
+            var expected = new byte[] { 0, 1, 2, 4, (byte)'a', (byte)'b', 0xfe, 0xff };
+            var ipfs = TestFixture.Ipfs;
+            var topic = "net-ipfs-http-client-test-" + Guid.NewGuid().ToString();
+            var cs = new CancellationTokenSource();
+            try
+            {
+                await ipfs.PubSub.SubscribeAsync(topic, msg =>
+                {
+                    messages.Add(msg);
+                }, cs.Token);
+                var ms = new MemoryStream(expected, false);
+                await ipfs.PubSub.PublishAsync(topic, ms);
+
+                await Task.Delay(1000);
+                Assert.AreEqual(1, messages.Count);
+                CollectionAssert.AreEqual(expected, messages[0].DataBytes);
+            }
+            finally
+            {
+                cs.Cancel();
+            }
+        }
     }
 }