feat(PubSubApi): publish a binary message

Author: richardschneider

src/CoreApi/PubSubApi.cs

@@ -42,6 +42,23 @@ internal PubSubApi(IpfsClient ipfs)
             return strings.Select(s => new Peer { Id = (string)s } );
         }
 
+        public Task PublishAsync(string topic, byte[] message, CancellationToken cancel = default(CancellationToken))
+        {
+            var url = new StringBuilder();
+            url.Append("/api/v0/pubsub/pub");
+            url.Append("?arg=");
+            url.Append(System.Net.WebUtility.UrlEncode(topic));
+            url.Append("&arg=");
+            var data = Encoding.ASCII.GetString(System.Net.WebUtility.UrlEncodeToBytes(message, 0, message.Length));
+            url.Append(data);
+            return ipfs.DoCommandAsync(new Uri(ipfs.ApiUri, url.ToString()), cancel);
+        }
+
+        public Task PublishAsync(string topic, Stream message, CancellationToken cancel = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
         public async Task PublishAsync(string topic, string message, CancellationToken cancel = default(CancellationToken))
         {
             var _ = await ipfs.DoCommandAsync("pubsub/pub", cancel, topic, "arg=" + message);
@@ -102,15 +119,6 @@ void ProcessMessages(string topic, Action<PublishedMessage> handler, StreamReade
             log.DebugFormat("Stop listening for '{0}' messages", topic);
         }
 
-        public Task PublishAsync(string topic, byte[] message, CancellationToken cancel = default(CancellationToken))
-        {
-            throw new NotImplementedException();
-        }
-
-        public Task PublishAsync(string topic, Stream message, CancellationToken cancel = default(CancellationToken))
-        {
-            throw new NotImplementedException();
-        }
     }
 
 }

src/IpfsClient.cs

@@ -272,6 +272,21 @@ public async Task<string> DoCommandAsync(string command, CancellationToken cance
             }
         }
 
+        internal async Task DoCommandAsync(Uri url, CancellationToken cancel)
+        {
+            if (log.IsDebugEnabled)
+                log.Debug("POST " + url.ToString());
+            using (var response = await Api().PostAsync(url, null, cancel))
+            {
+                await ThrowOnErrorAsync(response);
+                var body = await response.Content.ReadAsStringAsync();
+                if (log.IsDebugEnabled)
+                    log.Debug("RSP " + body);
+                return;
+            }
+        }
+
+
         /// <summary>
         ///   Perform an <see href="https://ipfs.io/docs/api/">IPFS API command</see> returning 
         ///   a specific <see cref="Type"/>.

test/CoreApi/PubSubApiTest.cs

@@ -2,6 +2,7 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using Newtonsoft.Json.Linq;
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -173,5 +174,31 @@ await ipfs.PubSub.SubscribeAsync(topic, msg =>
             await Task.Delay(1000);
             Assert.AreEqual(1, messageCount1);
         }
+
+        [TestMethod]
+        public async Task Subscribe_BinaryMessage()
+        {
+            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);
+                await ipfs.PubSub.PublishAsync(topic, expected);
+
+                await Task.Delay(1000);
+                Assert.AreEqual(1, messages.Count);
+                CollectionAssert.AreEqual(expected, messages[0].DataBytes);
+            }
+            finally
+            {
+                cs.Cancel();
+            }
+        }
     }
 }