use super::{TestResult, config, require, required};
use serde_json::{Value, json};
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
pub(super) struct GoWorker {
child: Option<Child>,
input: Option<ChildStdin>,
output: BufReader<ChildStdout>,
}
impl GoWorker {
pub(super) fn start(
binary: &Path,
subject: &str,
output_schema: &str,
) -> Result<Self, Box<dyn std::error::Error>> {
let config = config("BUILDER")?;
let output_schema: Value = serde_json::from_str(output_schema)?;
let mut child = Command::new(binary)
.env_remove("LIMINAL_SERVICE")
.env_remove("LIMINAL_TOKEN")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()?;
let input = child.stdin.take().ok_or("Go worker stdin absent")?;
let output = BufReader::new(child.stdout.take().ok_or("Go worker stdout absent")?);
let mut worker = Self {
child: Some(child),
input: Some(input),
output,
};
let command = json!({"version":1,"operation":"worker","address":config.server,
"pass_file":required("AGT003_BUILDER_PASS_FILE")?,"subject":subject,"behavior":"echo",
"input_schema":{"type":"object"},"output_schema":output_schema});
worker.send(&command)?;
Ok(worker)
}
fn send(&mut self, value: &Value) -> TestResult {
let input = self.input.as_mut().ok_or("worker input closed")?;
serde_json::to_writer(&mut *input, value)?;
input.write_all(b"\n")?;
input.flush()?;
Ok(())
}
pub(super) fn signal(&mut self) -> Result<Value, Box<dyn std::error::Error>> {
let bound = config("BUILDER")?.limits.record_bytes;
let mut bytes = Vec::new();
let count = std::io::Read::by_ref(&mut self.output)
.take(u64::try_from(bound)?.saturating_add(1))
.read_until(b'\n', &mut bytes)?;
require(
count > 0 && count <= bound && bytes.last() == Some(&b'\n'),
"Go worker did not return a bounded signal",
)?;
serde_json::from_slice(&bytes).map_err(|_| "Go worker signal is not JSON".into())
}
pub(super) fn stop(mut self) -> TestResult {
self.send(&json!({"version":1,"operation":"stop"}))?;
require(
self.signal()?["event"] == "stopped",
"Go worker did not signal its closed connection",
)?;
drop(self.input.take());
require(
self.child
.take()
.ok_or("Go worker absent")?
.wait()?
.success(),
"Go worker exited unsuccessfully",
)
}
}
impl Drop for GoWorker {
fn drop(&mut self) {
if let Some(mut child) = self.child.take() {
if let Err(error) = child.kill() {
eprintln!("owned Go worker stop failed: {}", error.kind());
}
if let Err(error) = child.wait() {
eprintln!("owned Go worker join failed: {}", error.kind());
}
}
}
}
pub(super) fn build_worker(
root: &Path,
) -> Result<(tempfile::TempDir, PathBuf), Box<dyn std::error::Error>> {
let directory = tempfile::TempDir::new_in(root)?;
let sdk = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../sdks/liminal-go")
.canonicalize()?;
let original = std::fs::read_to_string(sdk.join("go.mod"))?;
let go = original
.lines()
.find(|line| line.starts_with("go "))
.ok_or("Go language pin absent")?;
let toolchain = original
.lines()
.find(|line| line.starts_with("toolchain "))
.ok_or("Go toolchain pin absent")?;
std::fs::write(
directory.path().join("go.mod"),
format!(
"module sitefixture\n{go}\n{toolchain}\nrequire github.com/ablative-io/liminal/sdks/liminal-go v0.0.0\nreplace github.com/ablative-io/liminal/sdks/liminal-go => {}\n",
serde_json::to_string(&sdk)?
),
)?;
std::fs::write(directory.path().join("main.go"), GO_WORKER)?;
let binary = directory.path().join("worker");
let status = Command::new("go")
.args(["build", "-o"])
.arg(&binary)
.arg(".")
.current_dir(directory.path())
.env("GOCACHE", directory.path().join("cache"))
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()?;
require(
status.success(),
"the pinned Go SDK conformance worker did not build",
)?;
Ok((directory, binary))
}
const GO_WORKER: &str = r#"package main
import (
"context"
"errors"
"os"
"os/signal"
"syscall"
"github.com/ablative-io/liminal/sdks/liminal-go/conformance"
"github.com/ablative-io/liminal/sdks/liminal-go/service"
)
func run() (result error) {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
if os.Getenv("LIMINAL_SERVICE") == "" { return conformance.Run(ctx,os.Stdin,os.Stdout) }
worker, err := service.DialWorkerTCP(ctx,os.Getenv("LIMINAL_SERVER"),[]byte(os.Getenv("LIMINAL_TOKEN")),
service.Registration(os.Getenv("LIMINAL_SUBJECTS"),os.Getenv("LIMINAL_SERVICE"),`{"type":"object"}`,`{"type":"object"}`))
if err != nil { return err }
defer func(){ result=errors.Join(result,worker.Close()) }()
for { request, err := worker.Next(ctx); if err != nil { if ctx.Err()!=nil { return nil }; return err }
if err = request.Reply(ctx,request.Payload); err != nil { return err }
}
}
func main() { if run()!=nil { os.Exit(1) } }
"#;