Add multiplex workers support (detekt#46)

.github/workflows/main.yml

@@ -27,6 +27,8 @@ jobs:
         run: |
           curl --location --fail https://github.com/bazelbuild/bazelisk/releases/download/v1.1.0/bazelisk-linux-amd64 --output /tmp/bazelisk
           chmod +x /tmp/bazelisk && echo "::add-path::/tmp/"
+      - name: "Unit tests"
+        run: bazelisk test //detekt/wrapper:tests
       - name: "Analysis tests"
         run: bazelisk test //tests/analysis:tests
       - name: "Integration tests"

detekt/defs.bzl

@@ -69,6 +69,7 @@ def _impl(ctx):
         executable = ctx.executable._detekt_wrapper,
         execution_requirements = {
             "supports-workers": "1",
+            "supports-multiplex-workers": "1",
         },
         arguments = [java_arguments, detekt_arguments],
     )

detekt/toolchains.bzl

@@ -8,7 +8,7 @@ load("@rules_java//java:repositories.bzl", "rules_java_dependencies", "rules_jav
 load("@rules_jvm_external//:defs.bzl", "maven_install")
 load("@rules_proto//proto:repositories.bzl", "rules_proto_dependencies", "rules_proto_toolchains")
 
-def rules_detekt_toolchains(detekt_version = "1.2.2", toolchain = "@rules_detekt//detekt:default_toolchain"):
+def rules_detekt_toolchains(detekt_version = "1.3.0", toolchain = "@rules_detekt//detekt:default_toolchain"):
     """Invokes `rules_detekt` toolchains.
 
     Declares toolchains that are dependencies of the `rules_detekt` workspace.
@@ -34,6 +34,8 @@ def rules_detekt_toolchains(detekt_version = "1.2.2", toolchain = "@rules_detekt
         name = "rules_detekt_dependencies",
         artifacts = [
             "io.gitlab.arturbosch.detekt:detekt-cli:{v}".format(v = detekt_version),
+            "io.reactivex.rxjava2:rxjava:2.2.16",
+            "junit:junit:4.12",
         ],
         repositories = [
             "https://repo1.maven.org/maven2",

detekt/wrapper/BUILD

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_kotlin//kotlin:kotlin.bzl", "kt_jvm_library")
+load("@io_bazel_rules_kotlin//kotlin:kotlin.bzl", "kt_jvm_library", "kt_jvm_test")
 load("@rules_java//java:defs.bzl", "java_binary", "java_proto_library")
 load("@rules_proto//proto:defs.bzl", "proto_library")
 
@@ -18,6 +18,7 @@ kt_jvm_library(
     deps = [
         ":worker_protocol_java_proto",
         "@rules_detekt_dependencies//:io_gitlab_arturbosch_detekt_detekt_cli",
+        "@rules_detekt_dependencies//:io_reactivex_rxjava2_rxjava",
     ],
 )
 
@@ -27,3 +28,17 @@ java_binary(
     visibility = ["//visibility:public"],
     runtime_deps = [":lib"],
 )
+
+kt_jvm_test(
+    name = "tests",
+    srcs = glob([
+        "src/test/kotlin/**/*.kt",
+        "src/testFixtures/kotlin/**/*.kt",
+    ]),
+    friends = [":lib"],
+    test_class = "io.buildfoundation.bazel.rulesdetekt.wrapper.TestsSuite",
+    deps = [
+        ":lib",
+        "@rules_detekt_dependencies//:junit_junit",
+    ],
+)

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Application.kt

@@ -1,44 +1,42 @@
 package io.buildfoundation.bazel.rulesdetekt.wrapper
 
-import bazel.worker.WorkerProtocol
-import kotlin.system.exitProcess
-
-interface Application {
-
-    fun run(arguments: Array<String>)
-
-    class Worker(private val executor: SandboxExecutor) : Application {
-
-        override fun run(arguments: Array<String>) {
-            while (true) {
-                val result = WorkerProtocol.WorkRequest.parseDelimitedFrom(System.`in`).let {
-                    executor.execute(it.argumentsList.toTypedArray())
-                }
-
-                WorkerProtocol.WorkResponse.newBuilder()
-                        .apply {
-                            if (result.code != 0) {
-                                output = listOf(result.stdout, result.stderr).joinToString(separator = "\n")
-                            }
-                        }
-                        .setExitCode(result.code)
-                        .build()
-                        .writeDelimitedTo(System.out)
-            }
-        }
-    }
+import io.reactivex.Scheduler
+
+internal interface Application {
 
-    class OneShot(private val executor: SandboxExecutor) : Application {
+    fun run(args: Array<String>)
 
-        override fun run(arguments: Array<String>) {
-            val result = executor.execute(arguments)
+    class OneShot(
+            private val executable: Executable,
+            private val streams: Streams,
+            private val platform: Platform
+    ) : Application {
 
-            if (result.code != 0) {
-                System.out.println(result.stdout)
-                System.err.println(result.stderr)
+        override fun run(args: Array<String>) {
+            val result = executable.execute(args)
+
+            if (result is Result.Failure) {
+                streams.error.println(result.description)
             }
 
-            exitProcess(result.code)
+            platform.exit(result.consoleStatusCode)
+        }
+    }
+
+    class Worker(
+            private val scheduler: Scheduler,
+            private val executable: WorkerExecutable,
+            private val streams: WorkerStreams
+    ) : Application {
+
+        override fun run(args: Array<String>) {
+            streams.request
+                    .subscribeOn(scheduler)
+                    .parallel()
+                    .runOn(scheduler)
+                    .map { executable.execute(it) }
+                    .sequential()
+                    .blockingSubscribe(streams.response)
         }
     }
 }

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Detekt.kt

@@ -0,0 +1,16 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+import io.gitlab.arturbosch.detekt.cli.buildRunner
+import java.io.PrintStream
+
+internal interface Detekt {
+
+    fun execute(args: Array<String>, outputPrinter: PrintStream, errorPrinter: PrintStream)
+
+    class Impl : Detekt {
+
+        override fun execute(args: Array<String>, outputPrinter: PrintStream, errorPrinter: PrintStream) {
+            buildRunner(args, outputPrinter, errorPrinter).execute()
+        }
+    }
+}

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Executable.kt

@@ -0,0 +1,34 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+import java.io.ByteArrayOutputStream
+import java.io.PrintStream
+import java.nio.charset.Charset
+
+internal interface Executable {
+
+    fun execute(args: Array<String>): Result
+
+    class DetektImpl(private val detekt: Detekt) : Executable {
+
+        override fun execute(args: Array<String>): Result {
+            val outputPrinter = PrintStream(Streams.DevNullOutputStream().buffered())
+
+            val errorPrinterBuffer = ByteArrayOutputStream()
+            val errorPrinter = PrintStream(errorPrinterBuffer.buffered())
+
+            return try {
+                detekt.execute(args, outputPrinter, errorPrinter)
+
+                Result.Success
+            } catch (e: Exception) {
+                e.printStackTrace(errorPrinter)
+                errorPrinter.flush()
+
+                Result.Failure(errorPrinterBuffer.toString(Charset.defaultCharset()))
+            } finally {
+                outputPrinter.close()
+                errorPrinter.close()
+            }
+        }
+    }
+}

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Platform.kt

@@ -0,0 +1,13 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+import kotlin.system.exitProcess
+
+internal interface Platform {
+
+    fun exit(code: Int)
+
+    class Impl : Platform {
+
+        override fun exit(code: Int) = exitProcess(code)
+    }
+}

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Result.kt

@@ -0,0 +1,11 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+internal sealed class Result {
+    object Success : Result()
+    data class Failure(val description: String) : Result()
+
+    val consoleStatusCode = when (this) {
+        is Success -> 0
+        is Failure -> 1
+    }
+}

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/Streams.kt

@@ -0,0 +1,25 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+import java.io.InputStream
+import java.io.OutputStream
+import java.io.PrintStream
+
+internal data class Streams(val input: InputStream, val output: PrintStream, val error: PrintStream) {
+
+    companion object {
+        fun system(): Streams {
+            val systemInput = System.`in`
+            val systemOutput = System.out
+            val systemError = System.err
+
+            // Suppressing system singleton stream to avoid accidental output from the tool.
+            System.setOut(PrintStream(DevNullOutputStream()))
+
+            return Streams(systemInput, systemOutput, systemError)
+        }
+    }
+
+    class DevNullOutputStream : OutputStream() {
+        override fun write(byte: Int) = Unit
+    }
+}

detekt/wrapper/src/main/kotlin/io/buildfoundation/bazel/rulesdetekt/wrapper/WorkerExecutable.kt

@@ -0,0 +1,27 @@
+package io.buildfoundation.bazel.rulesdetekt.wrapper
+
+import bazel.worker.WorkerProtocol.WorkRequest
+import bazel.worker.WorkerProtocol.WorkResponse
+
+internal interface WorkerExecutable {
+
+    fun execute(request: WorkRequest): WorkResponse
+
+    class Impl(private val executable: Executable) : WorkerExecutable {
+
+        override fun execute(request: WorkRequest): WorkResponse {
+            val result = executable.execute(request.argumentsList.toTypedArray())
+
+            val resultOutput = when (result) {
+                is Result.Success -> WorkResponse.getDefaultInstance().output
+                is Result.Failure -> result.description
+            }
+
+            return WorkResponse.newBuilder()
+                    .setRequestId(request.requestId)
+                    .setOutput(resultOutput)
+                    .setExitCode(result.consoleStatusCode)
+                    .build()
+        }
+    }
+}