feat: send a request to a remote REST service and capture response (#52)

Streamistry.Core/Pipes/RestResponder.cs

@@ -0,0 +1,61 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using System.Linq;
+using System.Net.Http.Json;
+using System.Text;
+using System.Text.Json.Nodes;
+using System.Threading.Tasks;
+using Streamistry.Observability;
+using Streamistry.Pipes.Parsers;
+
+namespace Streamistry.Pipes;
+public class RestResponder<TInput, TOutput> : EscapeRouterPipe<TInput, TOutput>, IProcessablePipe<TInput?> where TOutput : JsonNode
+{
+    protected HttpClient Client { get; }
+    protected Func<TInput?, string> UrlBuiler { get; }
+
+    public RestResponder(IChainablePort<TInput> upstream, HttpClient client, Func<TInput?, string> urlBuilder)
+        : base(upstream)
+    {
+        Client = client;
+        UrlBuiler = urlBuilder;
+    }
+
+    [Meter]
+    protected override bool TryInvoke(TInput? obj, [NotNullWhen(true)] out TOutput? value)
+    {
+        value = null;
+        var url = UrlBuiler.Invoke(obj);
+        var response = RemoteInvoke(url);
+        if (response != null && response.StatusCode == System.Net.HttpStatusCode.OK)
+        {
+            var content = ConvertHttpContentToString(response.Content);
+            value = (TOutput?)JsonNode.Parse(content ?? string.Empty);
+            if (value is not null)
+                return true;
+        }
+        return false;
+    }
+
+    [Meter]
+    [Trace]
+    protected virtual HttpResponseMessage RemoteInvoke(string url)
+    {
+        var response = Client.Send(new HttpRequestMessage(HttpMethod.Get, url));
+        return response;
+    }
+
+    [Trace]
+    protected virtual string ConvertHttpContentToString(HttpContent content)
+    {
+        using (var stream = content.ReadAsStream())
+        using (var reader = new StreamReader(stream, Encoding.UTF8))
+        {
+            return reader.ReadToEnd();
+        }
+    }
+
+
+
+}

Streamistry.Testing/RestResponderTests.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Text.Json.Nodes;
+using System.Threading.Tasks;
+using Moq;
+using RichardSzalay.MockHttp;
+using NUnit.Framework;
+using Streamistry.Pipes;
+using Streamistry.Pipes.Mappers;
+using Streamistry.Pipes.Sinks;
+
+namespace Streamistry.Testing;
+public class RestResponderTests
+{
+    [Test]
+    public void Emit_ValidData_CallHttpClient()
+    {
+        var mockHttp = new MockHttpMessageHandler();
+        mockHttp.When("/customer/1").Respond("application/json", JsonTests.JsonFirst);
+        mockHttp.When("/customer/2").Respond("application/json", JsonTests.JsonSecond);
+        mockHttp.When("/customer/3").Respond("application/json", JsonTests.JsonThird);
+        var client = mockHttp.ToHttpClient();
+        client.BaseAddress = new Uri("http://localhost");
+
+        var pipeline = new Pipeline<int>();
+        var mapper = new RestResponder<int, JsonObject>(pipeline, client, x => $"/customer/{x}");
+        var plunker = new JsonPathPlucker<string>(mapper, "$.user.name");
+        var sink = new MemorySink<string>(plunker);
+
+        pipeline.Emit(1);
+        pipeline.Emit(2);
+        pipeline.Emit(3);
+
+        Assert.That(sink.State.Count, Is.EqualTo(3));
+        Assert.Multiple(() =>
+        {
+            Assert.That(sink.State.First(), Is.EqualTo("Albert Einstein"));
+            Assert.That(sink.State.Last(), Is.EqualTo("John von Neumann"));
+        });
+    }
+}

Streamistry.Testing/Streamistry.Testing.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
   <ItemGroup>
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.11.0" />
     <PackageReference Include="NUnit" Version="4.2.2" />
@@ -19,6 +19,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
+    <PackageReference Include="RichardSzalay.MockHttp" Version="7.0.0" />
   </ItemGroup>
   <ItemGroup>
 		<ProjectReference Include="..\..\Streamistry\Streamistry.Core\Streamistry.csproj" />