feat(manualsync): add dual-protocol mock server
- expose manual sync and health-check endpoints on ports 9001 and 9002 - add structured request logging and graceful shutdown - cover routing, validation, listener cleanup, and shutdown behavior - move async task API documentation into docs
This commit is contained in:
parent
2fbef3e1fa
commit
a825314737
|
|
@ -30,6 +30,7 @@ go.work
|
|||
/configs/**/*.pem
|
||||
|
||||
# ai config
|
||||
.agents/
|
||||
.cursor/
|
||||
.claude/
|
||||
.codewhale/
|
||||
|
|
@ -41,3 +42,6 @@ go.work
|
|||
ai-debug.log
|
||||
*.patch
|
||||
*.diff
|
||||
docs/agents/
|
||||
AGENTS.md
|
||||
skills-lock.json
|
||||
|
|
|
|||
|
|
@ -0,0 +1,22 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
func main() {
|
||||
logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
|
||||
err := run(ctx, logger, net.Listen)
|
||||
stop()
|
||||
if err != nil {
|
||||
logger.Error("manual sync mock stopped", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,168 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"modelRT/client/manualsync"
|
||||
)
|
||||
|
||||
type listenFunc func(network, address string) (net.Listener, error)
|
||||
|
||||
const (
|
||||
cl3611Address = ":9001"
|
||||
protocol104Address = ":9002"
|
||||
manualAPIPath = "/api/manual"
|
||||
healthPath = "/healthz"
|
||||
maxRequestBody = 1 << 20
|
||||
shutdownTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
func newProtocolServer(protocol, address string, logger *slog.Logger) *http.Server {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("POST "+manualAPIPath, manualHandler(protocol, strings.TrimPrefix(address, ":"), logger))
|
||||
mux.HandleFunc("GET "+healthPath, healthHandler)
|
||||
|
||||
return &http.Server{
|
||||
Addr: address,
|
||||
Handler: mux,
|
||||
ReadHeaderTimeout: 5 * time.Second,
|
||||
ReadTimeout: 10 * time.Second,
|
||||
WriteTimeout: 10 * time.Second,
|
||||
IdleTimeout: 60 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context, logger *slog.Logger, listen listenFunc) error {
|
||||
cl3611Listener, err := listen("tcp", cl3611Address)
|
||||
if err != nil {
|
||||
return fmt.Errorf("listen on %s: %w", cl3611Address, err)
|
||||
}
|
||||
protocol104Listener, err := listen("tcp", protocol104Address)
|
||||
if err != nil {
|
||||
_ = cl3611Listener.Close()
|
||||
return fmt.Errorf("listen on %s: %w", protocol104Address, err)
|
||||
}
|
||||
|
||||
servers := []struct {
|
||||
protocol string
|
||||
server *http.Server
|
||||
listener net.Listener
|
||||
}{
|
||||
{protocol: "cl3611", server: newProtocolServer("cl3611", cl3611Address, logger), listener: cl3611Listener},
|
||||
{protocol: "104", server: newProtocolServer("104", protocol104Address, logger), listener: protocol104Listener},
|
||||
}
|
||||
|
||||
serveErrors := make(chan error, len(servers))
|
||||
var serversWaitGroup sync.WaitGroup
|
||||
for _, configuredServer := range servers {
|
||||
serversWaitGroup.Add(1)
|
||||
go func() {
|
||||
defer serversWaitGroup.Done()
|
||||
logger.Info("manual sync mock server started",
|
||||
"protocol", configuredServer.protocol,
|
||||
"listen_addr", configuredServer.listener.Addr().String(),
|
||||
)
|
||||
if err := configuredServer.server.Serve(configuredServer.listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
serveErrors <- fmt.Errorf("%s server failed: %w", configuredServer.protocol, err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
var runError error
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case runError = <-serveErrors:
|
||||
}
|
||||
|
||||
shutdownContext, cancelShutdown := context.WithTimeout(context.Background(), shutdownTimeout)
|
||||
defer cancelShutdown()
|
||||
for _, configuredServer := range servers {
|
||||
if err := configuredServer.server.Shutdown(shutdownContext); err != nil {
|
||||
runError = errors.Join(runError, fmt.Errorf("shut down %s server: %w", configuredServer.protocol, err))
|
||||
_ = configuredServer.server.Close()
|
||||
}
|
||||
}
|
||||
serversWaitGroup.Wait()
|
||||
return runError
|
||||
}
|
||||
|
||||
func healthHandler(response http.ResponseWriter, _ *http.Request) {
|
||||
response.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
||||
response.WriteHeader(http.StatusOK)
|
||||
_, _ = response.Write([]byte("ok\n"))
|
||||
}
|
||||
|
||||
func manualHandler(protocol, listenPort string, logger *slog.Logger) http.HandlerFunc {
|
||||
return func(response http.ResponseWriter, request *http.Request) {
|
||||
request.Body = http.MaxBytesReader(response, request.Body, maxRequestBody)
|
||||
var payload manualsync.Request
|
||||
decoder := json.NewDecoder(request.Body)
|
||||
if err := decoder.Decode(&payload); err != nil {
|
||||
logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
|
||||
slog.String("protocol", protocol),
|
||||
slog.String("listen_port", listenPort),
|
||||
slog.String("remote_addr", request.RemoteAddr),
|
||||
slog.Any("error", err),
|
||||
)
|
||||
var maxBytesError *http.MaxBytesError
|
||||
if errors.As(err, &maxBytesError) {
|
||||
http.Error(response, "request body too large", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
http.Error(response, "invalid JSON request body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
var trailing any
|
||||
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
|
||||
if err == nil {
|
||||
err = errors.New("request body contains multiple JSON documents")
|
||||
}
|
||||
logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
|
||||
slog.String("protocol", protocol),
|
||||
slog.String("listen_port", listenPort),
|
||||
slog.String("remote_addr", request.RemoteAddr),
|
||||
slog.Any("error", err),
|
||||
)
|
||||
http.Error(response, "invalid JSON request body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
logger.LogAttrs(request.Context(), slog.LevelInfo, "manual sync request received",
|
||||
slog.String("protocol", protocol),
|
||||
slog.String("listen_port", listenPort),
|
||||
slog.String("remote_addr", request.RemoteAddr),
|
||||
slog.Int("mode", int(payload.Mode)),
|
||||
slog.Group("target",
|
||||
slog.Int("type", payload.Target.Type),
|
||||
slog.String("station", payload.Target.Station),
|
||||
slog.String("main_pos", payload.Target.MainPos),
|
||||
slog.String("sub_pos", payload.Target.SubPos),
|
||||
slog.String("option", payload.Target.Option),
|
||||
),
|
||||
slog.Attr{Key: "data", Value: samplesLogValue(payload.Data)},
|
||||
)
|
||||
response.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
}
|
||||
|
||||
func samplesLogValue(samples []manualsync.SyntheticData) slog.Value {
|
||||
attributes := make([]slog.Attr, 0, len(samples))
|
||||
for index, sample := range samples {
|
||||
attributes = append(attributes, slog.Group(strconv.Itoa(index),
|
||||
slog.Int64("time", sample.Time),
|
||||
slog.Float64("value", sample.Value),
|
||||
))
|
||||
}
|
||||
return slog.GroupValue(attributes...)
|
||||
}
|
||||
|
|
@ -0,0 +1,234 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestHealthEndpointReportsServerIsReady(t *testing.T) {
|
||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
||||
request := httptest.NewRequest(http.MethodGet, healthPath, nil)
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
result := response.Result()
|
||||
defer result.Body.Close()
|
||||
body, err := io.ReadAll(result.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("read response body: %v", err)
|
||||
}
|
||||
if result.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d", result.StatusCode, http.StatusOK)
|
||||
}
|
||||
if contentType := result.Header.Get("Content-Type"); contentType != "text/plain; charset=utf-8" {
|
||||
t.Errorf("Content-Type = %q, want %q", contentType, "text/plain; charset=utf-8")
|
||||
}
|
||||
if string(body) != "ok\n" {
|
||||
t.Errorf("body = %q, want %q", body, "ok\\n")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManualEndpointLogsStructuredRequest(t *testing.T) {
|
||||
var logs bytes.Buffer
|
||||
logger := slog.New(slog.NewTextHandler(&logs, nil))
|
||||
server := newProtocolServer("cl3611", cl3611Address, logger)
|
||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{
|
||||
"mode": 1,
|
||||
"data": [{"time": 1736305467506000000, "value": 1.25}],
|
||||
"target": {
|
||||
"type": 1,
|
||||
"station": "001",
|
||||
"main_pos": "ssu001",
|
||||
"sub_pos": "TM1",
|
||||
"option": "RMS"
|
||||
}
|
||||
}`))
|
||||
request.RemoteAddr = "127.0.0.1:52130"
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != http.StatusNoContent {
|
||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusNoContent)
|
||||
}
|
||||
for _, fragment := range []string{
|
||||
"protocol=cl3611",
|
||||
"listen_port=9001",
|
||||
"remote_addr=127.0.0.1:52130",
|
||||
"mode=1",
|
||||
"target.type=1",
|
||||
"target.station=001",
|
||||
"target.main_pos=ssu001",
|
||||
"target.sub_pos=TM1",
|
||||
"target.option=RMS",
|
||||
"time=1736305467506000000",
|
||||
"value=1.25",
|
||||
} {
|
||||
if !strings.Contains(logs.String(), fragment) {
|
||||
t.Errorf("log %q does not contain %q", logs.String(), fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManualEndpointRejectsInvalidJSONAndLogsTheError(t *testing.T) {
|
||||
var logs bytes.Buffer
|
||||
logger := slog.New(slog.NewTextHandler(&logs, nil))
|
||||
server := newProtocolServer("104", protocol104Address, logger)
|
||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":`))
|
||||
request.RemoteAddr = "127.0.0.1:52131"
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
|
||||
}
|
||||
for _, fragment := range []string{
|
||||
"level=WARN",
|
||||
"msg=\"invalid manual sync request\"",
|
||||
"protocol=104",
|
||||
"listen_port=9002",
|
||||
"remote_addr=127.0.0.1:52131",
|
||||
"error=",
|
||||
} {
|
||||
if !strings.Contains(logs.String(), fragment) {
|
||||
t.Errorf("log %q does not contain %q", logs.String(), fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManualEndpointRejectsOversizedRequest(t *testing.T) {
|
||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
||||
body := `{"extra":"` + strings.Repeat("a", maxRequestBody) + `"}`
|
||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(body))
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != http.StatusRequestEntityTooLarge {
|
||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusRequestEntityTooLarge)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManualEndpointRejectsMultipleJSONDocuments(t *testing.T) {
|
||||
server := newProtocolServer("104", protocol104Address, slog.Default())
|
||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":1}{"mode":0}`))
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunClosesFirstListenerWhenSecondListenerFails(t *testing.T) {
|
||||
rawListener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatalf("listen on temporary port: %v", err)
|
||||
}
|
||||
firstListener := rawListener.(*net.TCPListener)
|
||||
listenCalls := 0
|
||||
listen := func(_, _ string) (net.Listener, error) {
|
||||
listenCalls++
|
||||
if listenCalls == 1 {
|
||||
return firstListener, nil
|
||||
}
|
||||
return nil, errors.New("port is already in use")
|
||||
}
|
||||
|
||||
err = run(context.Background(), slog.Default(), listen)
|
||||
|
||||
if err == nil || !strings.Contains(err.Error(), protocol104Address) {
|
||||
t.Fatalf("run error = %v, want error containing %q", err, protocol104Address)
|
||||
}
|
||||
if err := firstListener.SetDeadline(time.Now()); err == nil {
|
||||
t.Fatal("first listener is still open")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunServesBothProtocolsAndStopsOnCancellation(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
addresses := make(chan string, 2)
|
||||
listen := func(_, _ string) (net.Listener, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err == nil {
|
||||
addresses <- listener.Addr().String()
|
||||
}
|
||||
return listener, err
|
||||
}
|
||||
runErrors := make(chan error, 1)
|
||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
go func() {
|
||||
runErrors <- run(ctx, logger, listen)
|
||||
}()
|
||||
|
||||
cl3611Addr := <-addresses
|
||||
protocol104Addr := <-addresses
|
||||
waitForHealthyEndpoint(t, cl3611Addr)
|
||||
waitForHealthyEndpoint(t, protocol104Addr)
|
||||
cancel()
|
||||
|
||||
select {
|
||||
case err := <-runErrors:
|
||||
if err != nil {
|
||||
t.Fatalf("run returned an error during graceful shutdown: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("servers did not stop after context cancellation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProtocolServerEnforcesMethodAndPathRouting(t *testing.T) {
|
||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
||||
tests := []struct {
|
||||
name string
|
||||
method string
|
||||
path string
|
||||
wantStatus int
|
||||
}{
|
||||
{name: "manual endpoint rejects GET", method: http.MethodGet, path: manualAPIPath, wantStatus: http.StatusMethodNotAllowed},
|
||||
{name: "unknown path is not found", method: http.MethodGet, path: "/unknown", wantStatus: http.StatusNotFound},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
request := httptest.NewRequest(test.method, test.path, nil)
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
server.Handler.ServeHTTP(response, request)
|
||||
|
||||
if response.Code != test.wantStatus {
|
||||
t.Fatalf("status = %d, want %d", response.Code, test.wantStatus)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func waitForHealthyEndpoint(t *testing.T, address string) {
|
||||
t.Helper()
|
||||
client := &http.Client{Timeout: 100 * time.Millisecond}
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
response, err := client.Get("http://" + address + healthPath)
|
||||
if err == nil {
|
||||
_ = response.Body.Close()
|
||||
if response.StatusCode == http.StatusOK {
|
||||
return
|
||||
}
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("health endpoint at %s did not become ready", address)
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue