feat: http fetch processor

build.gradle.kts

@@ -32,6 +32,7 @@ dependencies {
   implementation("io.ktor:ktor-client-cio:2.3.10")
   implementation("io.ktor:ktor-server-core:2.3.10")
   implementation("io.ktor:ktor-server-netty:2.3.10")
+  testImplementation("io.ktor:ktor-client-mock:2.3.10")
 
   // RDF dependencies.
   implementation("org.apache.jena:apache-jena-libs:5.0.0")

src/main/kotlin/std/HttpFetch.kt

@@ -0,0 +1,54 @@
+package technology.idlab.std
+
+import bridge.Writer
+import io.ktor.client.*
+import io.ktor.client.engine.*
+import io.ktor.client.engine.cio.*
+import io.ktor.client.request.*
+import io.ktor.client.statement.*
+import io.ktor.http.*
+import kotlinx.coroutines.runBlocking
+import technology.idlab.runner.Processor
+
+class HttpFetch(args: Map<String, Any>) : Processor(args) {
+  /** Meta configuration. */
+  private var engine: HttpClientEngine = CIO.create()
+
+  /** Parameters. */
+  private val endpoint = this.getArgument<String>("endpoint")
+  private val output = this.getArgument<Writer>("output")
+  private val headers = this.getArgument<Array<String>>("headers")
+  private val method = this.getNullableArgument<String>("method") ?: "GET"
+
+  /** Prebuild request. */
+  private val builder = HttpRequestBuilder()
+
+  /** Build the HTTP request. */
+  init {
+    builder.url(endpoint)
+    builder.method = HttpMethod.parse(method)
+    headers.map { header ->
+      val (key, value) = header.split(":").map { it.trim() }
+      builder.headers.append(key, value)
+    }
+  }
+
+  /** Execute an HTTP request and output it to the writer. */
+  override fun exec() = runBlocking {
+    val client = HttpClient(engine)
+    val res = client.request(builder)
+
+    // Check validity of result.
+    if (!res.status.isSuccess()) {
+      log.fatal("ERROR: Status code ${res.status.value} received from $endpoint")
+    }
+
+    // Push the result to the output.
+    val bytes = res.readBytes()
+    output.push(bytes)
+  }
+
+  internal fun overwriteEngine(engine: HttpClientEngine) {
+    this.engine = engine
+  }
+}

src/main/kotlin/std/README.md

@@ -1,15 +1,22 @@
-# Standard Runner Library
+# Standard Processor Library
 
 This directory contains implementations of processors which can be used directly by the end user, without the need for additional dependencies. These also exist as a reference to processor developers.
 
-Their RDF definitions can be found [here](../../resources/std).
+Their RDF definitions can be found [here](./../../resources/std). These are included in the default ontology.
+
+## Network Utilities
+
+These processors interact with the network.
+
+| Processor                     | Description                       |
+|-------------------------------|-----------------------------------|
+| [jvm:HttpFetch](HttpFetch.kt) | Reads data from an HTTP endpoint. |
 
 ## File Utilities
 
 These processors interact with the local file system.
 
-| Processor                   | Description                                                            |
-|-----------------------------|------------------------------------------------------------------------|
-| [FileReader](FileReader.kt) | Reads a file with a given `path` from the local file system.           |
-| [FileWriter](FileWriter.kt) | Overwrites/appends a file with a given `path` using the incoming data. |
-
+| Processor                       | Description                                                            |
+|---------------------------------|------------------------------------------------------------------------|
+| [jvm:FileReader](FileReader.kt) | Reads a file with a given `path` from the local file system.           |
+| [jvm:FileWriter](FileWriter.kt) | Overwrites/appends a file with a given `path` using the incoming data. |

src/main/resources/std/http_fetch.ttl

@@ -0,0 +1,38 @@
+@prefix jvm: <https://w3id.org/conn/jvm#>.
+@prefix owl: <http://www.w3.org/2002/07/owl#>.
+@prefix sh: <http://www.w3.org/ns/shacl#>.
+@prefix xsd: <http://www.w3.org/2001/XMLSchema#>.
+
+<> owl:imports <../pipeline.ttl>.
+
+jvm:HttpFetch a jvm:Processor;
+  jvm:file <../../kotlin/std/HttpFetch.kt>;
+  jvm:language "Kotlin".
+
+[] a sh:NodeShape;
+  sh:targetClass jvm:HttpFetch;
+  sh:property [
+    sh:path jvm:endpoint;
+    sh:name "endpoint";
+    sh:datatype xsd:string;
+    sh:minCount 1;
+    sh:maxCount 1;
+  ], [
+    sh:path jvm:output;
+    sh:name "output";
+    sh:class jvm:ChannelWriter;
+    sh:minCount 1;
+    sh:maxCount 1;
+  ], [
+    sh:path jvm:headers;
+    sh:name "headers";
+    sh:datatype xsd:string;
+  ], [
+    sh:path jvm:method;
+    sh:name "method";
+    sh:datatype xsd:string;
+    sh:in ("GET" "POST" "PUT" "DELETE" "PATCH");
+    sh:maxCount 1;
+  ];
+  sh:closed true;
+  sh:ignoredProperties (rdf:type).

src/test/kotlin/std/HttpFetchTest.kt

@@ -0,0 +1,59 @@
+package std
+
+import bridge.DummyWriter
+import io.ktor.client.engine.mock.*
+import io.ktor.http.*
+import io.ktor.utils.io.*
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import technology.idlab.std.HttpFetch
+
+class HttpFetchTest {
+  @Test
+  fun functionality() {
+    // Initialize the processor arguments.
+    val writer = DummyWriter()
+    val args =
+        mapOf(
+            "endpoint" to "http://localhost:8080",
+            "method" to "DELETE",
+            "output" to writer,
+            "headers" to
+                arrayOf(
+                    "Content-Type: text/plain",
+                    "Key: Value",
+                ),
+        )
+
+    // Mock the HTTP engine.
+    val engine = MockEngine { request ->
+      // Check URL.
+      assertEquals("http://localhost:8080", request.url.toString())
+
+      // Check method.
+      assertEquals(HttpMethod.Delete, request.method)
+
+      // Check headers.
+      assertEquals("text/plain", request.headers["Content-Type"])
+      assertEquals("Value", request.headers["Key"])
+
+      // Send response.
+      respond(
+          content = ByteReadChannel("Hello, World!"),
+          status = HttpStatusCode.OK,
+      )
+    }
+
+    // Initialize the processor.
+    val httpFetch = HttpFetch(args)
+    httpFetch.overwriteEngine(engine)
+
+    // Execute.
+    httpFetch.exec()
+    val results = writer.getValues()
+
+    // Check body of response.
+    assertEquals(1, results.size)
+    assertEquals("Hello, World!", String(results[0]))
+  }
+}