Merge pull request #52 from richardschneider/core-api-50-0

Upgrade to core api v0.50.0
richardschneider · May 10, 2019 · 0996705 · 0996705
2 parents 60e4b2f + d5da246
commit 0996705
Show file tree

Hide file tree

Showing 6 changed files with 119 additions and 7 deletions.
diff --git a/src/CoreApi/DhtApi.cs b/src/CoreApi/DhtApi.cs
@@ -29,12 +29,33 @@ public Task<Peer> FindPeerAsync(MultiHash id, CancellationToken cancel = default
             return ipfs.IdAsync(id, cancel);
         }
 
-        public async Task<IEnumerable<Peer>> FindProvidersAsync(Cid id, int limit = 20, CancellationToken cancel = default(CancellationToken))
-        {
+        public async Task<IEnumerable<Peer>> FindProvidersAsync(Cid id, int limit = 20, Action<Peer> providerFound = null, CancellationToken cancel = default(CancellationToken))
+        {
+            // TODO: providerFound action
             var stream = await ipfs.PostDownloadAsync("dht/findprovs", cancel, id, $"num-providers={limit}");
             return ProviderFromStream(stream, limit);
-        }
-
+        }
+
+        public Task<byte[]> GetAsync(byte[] key, CancellationToken cancel = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task ProvideAsync(Cid cid, bool advertise = true, CancellationToken cancel = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task PutAsync(byte[] key, out byte[] value, CancellationToken cancel = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task<bool> TryGetAsync(byte[] key, out byte[] value, CancellationToken cancel = default(CancellationToken))
+        {
+            throw new NotImplementedException();
+        }
+
         IEnumerable<Peer> ProviderFromStream(Stream stream, int limit = int.MaxValue)
         { 
             using (var sr = new StreamReader(stream))

diff --git a/src/CoreApi/PubSubApi.cs b/src/CoreApi/PubSubApi.cs
@@ -42,6 +42,27 @@ public async Task<IEnumerable<Peer>> PeersAsync(string topic = null, Cancellatio
             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))
+        {
+            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))
         {
             var _ = await ipfs.DoCommandAsync("pubsub/pub", cancel, topic, "arg=" + message);

diff --git a/src/IpfsClient.cs b/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"/>.

diff --git a/src/IpfsHttpClient.csproj b/src/IpfsHttpClient.csproj
@@ -27,8 +27,8 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Ipfs.Core" Version="0.40.0" />
-    <PackageReference Include="Newtonsoft.Json" Version="11.0.2" />
+    <PackageReference Include="Ipfs.Core" Version="0.50.0" />
+    <PackageReference Include="Newtonsoft.Json" Version="12.0.1" />
 	  <PackageReference Include="System.Net.Http" Version="4.3.3" Condition="'$(TargetFramework)' == 'netstandard14'" />
     <PackageReference Include="System.Net.Http" Version="4.3.3" Condition="'$(TargetFramework)' == 'net45'" />
   </ItemGroup>

diff --git a/test/CoreApi/DhtApiTest.cs b/test/CoreApi/DhtApiTest.cs
@@ -28,7 +28,7 @@ public async Task FindProviders()
         {
             var ipfs = TestFixture.Ipfs;
             var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
-            var providers = await ipfs.Dht.FindProvidersAsync(helloWorldID, 1, cts.Token);
+            var providers = await ipfs.Dht.FindProvidersAsync(helloWorldID, 1, cancel: cts.Token);
             Assert.AreNotEqual(0, providers.Count());
         }
 

diff --git a/test/CoreApi/PubSubApiTest.cs b/test/CoreApi/PubSubApiTest.cs
@@ -2,6 +2,8 @@
 using Microsoft.VisualStudio.TestTools.UnitTesting;
 using Newtonsoft.Json.Linq;
 using System;
+using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Text;
 using System.Threading;
@@ -173,5 +175,58 @@ public async Task Unsubscribe()
             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();
+            }
+        }
+
+        [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();
+            }
+        }
     }
 }