diff --git a/cmd/neocode-gateway/main.go b/cmd/neocode-gateway/main.go new file mode 100644 index 00000000..0e76fa83 --- /dev/null +++ b/cmd/neocode-gateway/main.go @@ -0,0 +1,42 @@ +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "os" + "os/signal" + "syscall" + + "neo-code/internal/gateway" + "neo-code/internal/gateway/adapters" +) + +// main 负责启动独立 Gateway 进程并监听本地 IPC 请求。 +func main() { + transportMode := flag.String("transport", "auto", "IPC transport mode: auto|npipe|uds") + endpoint := flag.String("endpoint", "", "IPC endpoint path (socket path or named pipe)") + flag.Parse() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + bootstrap, err := gateway.NewBootstrap(gateway.BootstrapConfig{ + Transport: *transportMode, + Endpoint: *endpoint, + }, adapters.NewCoreMock()) + if err != nil { + fmt.Fprintf(os.Stderr, "neocode-gateway: %v\n", err) + os.Exit(1) + } + defer func() { + _ = bootstrap.Close() + }() + + fmt.Fprintf(os.Stdout, "neocode-gateway listening on %s\n", bootstrap.Endpoint()) + if err := bootstrap.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + fmt.Fprintf(os.Stderr, "neocode-gateway: %v\n", err) + os.Exit(1) + } +} diff --git a/go.mod b/go.mod index 7d1726d8..0046f29c 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( github.com/muesli/cancelreader v0.2.2 // indirect github.com/muesli/reflow v0.3.0 // indirect github.com/muesli/termenv v0.16.0 // indirect + github.com/natefinch/npipe v0.0.0-20160621034901-c1b8fa8bdcce // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect diff --git a/go.sum b/go.sum index 56abd244..1f9326c7 100644 --- a/go.sum +++ b/go.sum @@ -88,6 +88,8 @@ github.com/muesli/reflow v0.3.0 h1:IFsN6K9NfGtjeggFP+68I4chLZV2yIKsXJFNZ+eWh6s= github.com/muesli/reflow v0.3.0/go.mod h1:pbwTDkVPibjO2kyvBQRBxTWEEGDGq0FlB1BIKtnHY/8= github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc= github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk= +github.com/natefinch/npipe v0.0.0-20160621034901-c1b8fa8bdcce h1:TqjP/BTDrwN7zP9xyXVuLsMBXYMt6LLYi55PlrIcq8U= +github.com/natefinch/npipe v0.0.0-20160621034901-c1b8fa8bdcce/go.mod h1:ifHPsLndGGzvgzcaXUvzmt6LxKT4pJ+uzEhtnMt+f7A= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/internal/gateway/adapters/core_mock.go b/internal/gateway/adapters/core_mock.go new file mode 100644 index 00000000..635e7e39 --- /dev/null +++ b/internal/gateway/adapters/core_mock.go @@ -0,0 +1,36 @@ +package adapters + +import "context" + +const defaultPongMessage = "Pong" + +// CoreClient 定义网关访问核心能力的最小端口接口。 +type CoreClient interface { + Ping(ctx context.Context) (string, error) +} + +// CoreMock 是 CoreClient 的最小可运行 mock 实现。 +type CoreMock struct { + pong string +} + +// NewCoreMock 创建默认返回 Pong 的核心 mock。 +func NewCoreMock() *CoreMock { + return &CoreMock{pong: defaultPongMessage} +} + +// NewCoreMockWithPong 创建可自定义 Pong 文案的核心 mock。 +func NewCoreMockWithPong(pong string) *CoreMock { + if pong == "" { + pong = defaultPongMessage + } + return &CoreMock{pong: pong} +} + +// Ping 返回 mock 端固定的 pong 结果。 +func (m *CoreMock) Ping(_ context.Context) (string, error) { + if m == nil { + return defaultPongMessage, nil + } + return m.pong, nil +} diff --git a/internal/gateway/adapters/core_mock_test.go b/internal/gateway/adapters/core_mock_test.go new file mode 100644 index 00000000..0c4e9cad --- /dev/null +++ b/internal/gateway/adapters/core_mock_test.go @@ -0,0 +1,39 @@ +package adapters + +import ( + "context" + "testing" +) + +func TestCoreMockPing(t *testing.T) { + mock := NewCoreMock() + pong, err := mock.Ping(context.Background()) + if err != nil { + t.Fatalf("Ping() error = %v", err) + } + if pong != "Pong" { + t.Fatalf("Ping() = %s, want Pong", pong) + } +} + +func TestCoreMockPingCustomMessage(t *testing.T) { + mock := NewCoreMockWithPong("Custom") + pong, err := mock.Ping(context.Background()) + if err != nil { + t.Fatalf("Ping() error = %v", err) + } + if pong != "Custom" { + t.Fatalf("Ping() = %s, want Custom", pong) + } +} + +func TestCoreMockPingNilReceiver(t *testing.T) { + var mock *CoreMock + pong, err := mock.Ping(context.Background()) + if err != nil { + t.Fatalf("Ping() error = %v", err) + } + if pong != "Pong" { + t.Fatalf("Ping() = %s, want Pong", pong) + } +} diff --git a/internal/gateway/bootstrap.go b/internal/gateway/bootstrap.go new file mode 100644 index 00000000..e1e2cec5 --- /dev/null +++ b/internal/gateway/bootstrap.go @@ -0,0 +1,69 @@ +package gateway + +import ( + "context" + "errors" + "fmt" + "strings" + + "neo-code/internal/gateway/adapters" + "neo-code/internal/gateway/handlers" + "neo-code/internal/gateway/protocol" + "neo-code/internal/gateway/transport" +) + +// BootstrapConfig 描述网关进程启动时的最小配置。 +type BootstrapConfig struct { + Transport string + Endpoint string +} + +// Bootstrap 负责装配并驱动 Gateway IPC 服务器。 +type Bootstrap struct { + server *transport.Server +} + +// NewBootstrap 根据配置与 Core 端口装配可运行的网关实例。 +func NewBootstrap(cfg BootstrapConfig, core adapters.CoreClient) (*Bootstrap, error) { + if core == nil { + core = adapters.NewCoreMock() + } + + router := protocol.NewRouter() + pingHandler := handlers.NewPingHandler(core) + router.Register(protocol.MethodCorePing, pingHandler.Handle) + + server, err := transport.NewServer(transport.Config{ + Mode: transport.Mode(strings.TrimSpace(cfg.Transport)), + Endpoint: strings.TrimSpace(cfg.Endpoint), + }, router.HandleRaw) + if err != nil { + return nil, fmt.Errorf("build gateway server: %w", err) + } + + return &Bootstrap{server: server}, nil +} + +// Endpoint 返回网关当前监听地址。 +func (b *Bootstrap) Endpoint() string { + if b == nil || b.server == nil { + return "" + } + return b.server.Endpoint() +} + +// Run 启动网关服务并在上下文取消后退出。 +func (b *Bootstrap) Run(ctx context.Context) error { + if b == nil || b.server == nil { + return errors.New("gateway bootstrap is not initialized") + } + return b.server.Serve(ctx) +} + +// Close 主动关闭网关服务监听器。 +func (b *Bootstrap) Close() error { + if b == nil || b.server == nil { + return nil + } + return b.server.Close() +} diff --git a/internal/gateway/bootstrap_test.go b/internal/gateway/bootstrap_test.go new file mode 100644 index 00000000..d33dc671 --- /dev/null +++ b/internal/gateway/bootstrap_test.go @@ -0,0 +1,87 @@ +package gateway + +import ( + "context" + "encoding/json" + "net" + "path/filepath" + "runtime" + "testing" + "time" +) + +func TestBootstrapPingPongOverIPC(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uds integration test runs on non-windows") + } + + socketPath := filepath.Join(t.TempDir(), "gateway.sock") + bootstrap, err := NewBootstrap(BootstrapConfig{ + Transport: "uds", + Endpoint: socketPath, + }, nil) + if err != nil { + t.Fatalf("NewBootstrap() error = %v", err) + } + defer func() { + _ = bootstrap.Close() + }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + runDone := make(chan error, 1) + go func() { + runDone <- bootstrap.Run(ctx) + }() + + conn := waitForGatewayDial(t, bootstrap.Endpoint()) + defer conn.Close() + + encoder := json.NewEncoder(conn) + decoder := json.NewDecoder(conn) + if err := encoder.Encode(map[string]any{ + "jsonrpc": "2.0", + "id": "1", + "method": "core.ping", + }); err != nil { + t.Fatalf("Encode() error = %v", err) + } + + var response map[string]any + if err := decoder.Decode(&response); err != nil { + t.Fatalf("Decode() error = %v", err) + } + + if response["error"] != nil { + t.Fatalf("response error = %v, want nil", response["error"]) + } + if got := response["result"]; got != "Pong" { + t.Fatalf("response result = %v, want Pong", got) + } + + cancel() + select { + case err := <-runDone: + if err != nil { + t.Fatalf("Run() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Run() did not exit after cancel") + } +} + +func waitForGatewayDial(t *testing.T, endpoint string) net.Conn { + t.Helper() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + conn, err := net.Dial("unix", endpoint) + if err == nil { + return conn + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("dial gateway endpoint %s timeout", endpoint) + return nil +} diff --git a/internal/gateway/handlers/ping.go b/internal/gateway/handlers/ping.go new file mode 100644 index 00000000..dca407bd --- /dev/null +++ b/internal/gateway/handlers/ping.go @@ -0,0 +1,32 @@ +package handlers + +import ( + "context" + + "neo-code/internal/gateway/adapters" + "neo-code/internal/gateway/protocol" +) + +// PingHandler 负责处理 core.ping 请求并返回基础连通性结果。 +type PingHandler struct { + core adapters.CoreClient +} + +// NewPingHandler 创建 ping 请求处理器。 +func NewPingHandler(core adapters.CoreClient) *PingHandler { + return &PingHandler{core: core} +} + +// Handle 执行 ping 请求并把结果映射为 JSON-RPC 响应。 +func (h *PingHandler) Handle(ctx context.Context, req protocol.Request) protocol.Response { + if h == nil || h.core == nil { + return protocol.NewErrorResponse(req.ID, protocol.ErrorCodeInternalError, "core client unavailable") + } + + pong, err := h.core.Ping(ctx) + if err != nil { + return protocol.NewErrorResponse(req.ID, protocol.ErrorCodeInternalError, "core ping failed") + } + + return protocol.NewSuccessResponse(req.ID, pong) +} diff --git a/internal/gateway/handlers/ping_test.go b/internal/gateway/handlers/ping_test.go new file mode 100644 index 00000000..80374554 --- /dev/null +++ b/internal/gateway/handlers/ping_test.go @@ -0,0 +1,57 @@ +package handlers + +import ( + "context" + "errors" + "testing" + + "neo-code/internal/gateway/protocol" +) + +type coreClientStub struct { + pong string + err error +} + +func (s coreClientStub) Ping(_ context.Context) (string, error) { + if s.err != nil { + return "", s.err + } + return s.pong, nil +} + +func TestPingHandlerHandleSuccess(t *testing.T) { + handler := NewPingHandler(coreClientStub{pong: "Pong"}) + response := handler.Handle(context.Background(), protocol.Request{ID: []byte(`"1"`)}) + + if response.Error != nil { + t.Fatalf("Handle() error = %+v, want nil", response.Error) + } + if response.Result != "Pong" { + t.Fatalf("Handle() result = %v, want Pong", response.Result) + } +} + +func TestPingHandlerHandleCoreError(t *testing.T) { + handler := NewPingHandler(coreClientStub{err: errors.New("boom")}) + response := handler.Handle(context.Background(), protocol.Request{ID: []byte(`"1"`)}) + + if response.Error == nil { + t.Fatal("Handle() error = nil, want internal error") + } + if response.Error.Code != protocol.ErrorCodeInternalError { + t.Fatalf("Handle() error code = %d, want %d", response.Error.Code, protocol.ErrorCodeInternalError) + } +} + +func TestPingHandlerHandleNilCore(t *testing.T) { + handler := NewPingHandler(nil) + response := handler.Handle(context.Background(), protocol.Request{}) + + if response.Error == nil { + t.Fatal("Handle() error = nil, want internal error") + } + if response.Error.Code != protocol.ErrorCodeInternalError { + t.Fatalf("Handle() error code = %d, want %d", response.Error.Code, protocol.ErrorCodeInternalError) + } +} diff --git a/internal/gateway/protocol/jsonrpc.go b/internal/gateway/protocol/jsonrpc.go new file mode 100644 index 00000000..933b4985 --- /dev/null +++ b/internal/gateway/protocol/jsonrpc.go @@ -0,0 +1,154 @@ +package protocol + +import ( + "context" + "encoding/json" + "fmt" + "strings" +) + +const ( + // Version 表示网关当前支持的 JSON-RPC 版本。 + Version = "2.0" + // MethodCorePing 表示 core ping 请求方法名。 + MethodCorePing = "core.ping" +) + +const ( + // ErrorCodeParseError 表示请求 JSON 解码失败。 + ErrorCodeParseError = -32700 + // ErrorCodeInvalidRequest 表示请求结构不合法。 + ErrorCodeInvalidRequest = -32600 + // ErrorCodeMethodNotFound 表示方法不存在。 + ErrorCodeMethodNotFound = -32601 + // ErrorCodeInternalError 表示处理流程内部错误。 + ErrorCodeInternalError = -32603 +) + +// Request 表示网关 JSON-RPC 请求结构。 +type Request struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method"` + Params json.RawMessage `json:"params,omitempty"` +} + +// Response 表示网关 JSON-RPC 响应结构。 +type Response struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id,omitempty"` + Result any `json:"result,omitempty"` + Error *ResponseError `json:"error,omitempty"` +} + +// ResponseError 表示 JSON-RPC 错误负载。 +type ResponseError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// MethodHandler 表示单个 JSON-RPC 方法处理器。 +type MethodHandler func(ctx context.Context, req Request) Response + +// Router 负责 JSON-RPC 的解码、路由与编码。 +type Router struct { + handlers map[string]MethodHandler +} + +// NewRouter 创建 JSON-RPC 路由器。 +func NewRouter() *Router { + return &Router{handlers: make(map[string]MethodHandler)} +} + +// Register 注册一个方法处理器。 +func (r *Router) Register(method string, handler MethodHandler) { + if r == nil || strings.TrimSpace(method) == "" || handler == nil { + return + } + r.handlers[method] = handler +} + +// HandleRaw 处理原始 JSON 请求并返回编码后的响应。 +func (r *Router) HandleRaw(ctx context.Context, payload []byte) []byte { + request, err := DecodeRequest(payload) + if err != nil { + return mustEncodeResponse(NewErrorResponse(nil, ErrorCodeParseError, "parse error")) + } + response := r.handle(ctx, request) + return mustEncodeResponse(response) +} + +// DecodeRequest 负责把原始 JSON 解码为请求对象并做最小校验。 +func DecodeRequest(payload []byte) (Request, error) { + var req Request + if err := json.Unmarshal(payload, &req); err != nil { + return Request{}, fmt.Errorf("decode request: %w", err) + } + if strings.TrimSpace(req.JSONRPC) == "" { + req.JSONRPC = Version + } + return req, nil +} + +// EncodeResponse 把响应对象编码为 JSON 字节。 +func EncodeResponse(resp Response) ([]byte, error) { + data, err := json.Marshal(resp) + if err != nil { + return nil, fmt.Errorf("encode response: %w", err) + } + return data, nil +} + +// NewSuccessResponse 构建标准成功响应。 +func NewSuccessResponse(id json.RawMessage, result any) Response { + return Response{ + JSONRPC: Version, + ID: id, + Result: result, + } +} + +// NewErrorResponse 构建标准错误响应。 +func NewErrorResponse(id json.RawMessage, code int, message string) Response { + return Response{ + JSONRPC: Version, + ID: id, + Error: &ResponseError{ + Code: code, + Message: message, + }, + } +} + +// handle 执行请求校验、方法路由与错误映射。 +func (r *Router) handle(ctx context.Context, request Request) Response { + if errResponse, ok := validateRequest(request); ok { + return errResponse + } + handler, ok := r.handlers[request.Method] + if !ok { + return NewErrorResponse(request.ID, ErrorCodeMethodNotFound, "method not found") + } + return handler(ctx, request) +} + +// validateRequest 校验请求中的协议版本与方法名字段。 +func validateRequest(request Request) (Response, bool) { + if strings.TrimSpace(request.JSONRPC) != Version { + return NewErrorResponse(request.ID, ErrorCodeInvalidRequest, "invalid jsonrpc version"), true + } + if strings.TrimSpace(request.Method) == "" { + return NewErrorResponse(request.ID, ErrorCodeInvalidRequest, "missing method"), true + } + return Response{}, false +} + +// mustEncodeResponse 在编码失败时回退为内部错误响应。 +func mustEncodeResponse(response Response) []byte { + data, err := EncodeResponse(response) + if err == nil { + return data + } + fallback, _ := EncodeResponse(NewErrorResponse(nil, ErrorCodeInternalError, "internal error")) + return fallback +} diff --git a/internal/gateway/protocol/jsonrpc_test.go b/internal/gateway/protocol/jsonrpc_test.go new file mode 100644 index 00000000..a3f1bd26 --- /dev/null +++ b/internal/gateway/protocol/jsonrpc_test.go @@ -0,0 +1,88 @@ +package protocol + +import ( + "context" + "encoding/json" + "testing" +) + +func TestHandleRawPingSuccess(t *testing.T) { + router := NewRouter() + router.Register(MethodCorePing, func(_ context.Context, req Request) Response { + return NewSuccessResponse(req.ID, "Pong") + }) + + payload := []byte(`{"jsonrpc":"2.0","id":"1","method":"core.ping"}`) + response := router.HandleRaw(context.Background(), payload) + + var decoded Response + if err := json.Unmarshal(response, &decoded); err != nil { + t.Fatalf("HandleRaw() decode response error = %v", err) + } + if decoded.Error != nil { + t.Fatalf("HandleRaw() error = %+v, want nil", decoded.Error) + } + if got := decoded.Result; got != "Pong" { + t.Fatalf("HandleRaw() result = %v, want Pong", got) + } +} + +func TestHandleRawParseError(t *testing.T) { + router := NewRouter() + response := router.HandleRaw(context.Background(), []byte(`{"jsonrpc":"2.0","method":`)) + + var decoded Response + if err := json.Unmarshal(response, &decoded); err != nil { + t.Fatalf("HandleRaw() decode response error = %v", err) + } + if decoded.Error == nil { + t.Fatal("HandleRaw() error = nil, want parse error") + } + if decoded.Error.Code != ErrorCodeParseError { + t.Fatalf("HandleRaw() error code = %d, want %d", decoded.Error.Code, ErrorCodeParseError) + } +} + +func TestHandleRawMethodNotFound(t *testing.T) { + router := NewRouter() + payload := []byte(`{"jsonrpc":"2.0","id":"1","method":"core.unknown"}`) + response := router.HandleRaw(context.Background(), payload) + + var decoded Response + if err := json.Unmarshal(response, &decoded); err != nil { + t.Fatalf("HandleRaw() decode response error = %v", err) + } + if decoded.Error == nil { + t.Fatal("HandleRaw() error = nil, want method not found") + } + if decoded.Error.Code != ErrorCodeMethodNotFound { + t.Fatalf("HandleRaw() error code = %d, want %d", decoded.Error.Code, ErrorCodeMethodNotFound) + } +} + +func TestHandleRawInvalidRequest(t *testing.T) { + router := NewRouter() + payload := []byte(`{"jsonrpc":"1.0","id":"1","method":"core.ping"}`) + response := router.HandleRaw(context.Background(), payload) + + var decoded Response + if err := json.Unmarshal(response, &decoded); err != nil { + t.Fatalf("HandleRaw() decode response error = %v", err) + } + if decoded.Error == nil { + t.Fatal("HandleRaw() error = nil, want invalid request") + } + if decoded.Error.Code != ErrorCodeInvalidRequest { + t.Fatalf("HandleRaw() error code = %d, want %d", decoded.Error.Code, ErrorCodeInvalidRequest) + } +} + +func TestDecodeRequestDefaultVersion(t *testing.T) { + decoded, err := DecodeRequest([]byte(`{"method":"core.ping"}`)) + if err != nil { + t.Fatalf("DecodeRequest() error = %v", err) + } + if decoded.JSONRPC != Version { + t.Fatalf("DecodeRequest() jsonrpc = %s, want %s", decoded.JSONRPC, Version) + } +} diff --git a/internal/gateway/transport/npipe_windows.go b/internal/gateway/transport/npipe_windows.go new file mode 100644 index 00000000..69e5bffd --- /dev/null +++ b/internal/gateway/transport/npipe_windows.go @@ -0,0 +1,31 @@ +//go:build windows + +package transport + +import ( + "fmt" + "net" + "strings" + + "github.com/natefinch/npipe" +) + +const defaultPipeName = `\\.\pipe\neocode-gateway` + +// listenNPipe 在 Windows 平台创建 Named Pipe 监听器。 +func listenNPipe(endpoint string) (net.Listener, string, error) { + pipeAddress := strings.TrimSpace(endpoint) + if pipeAddress == "" { + pipeAddress = defaultPipeName + } + listener, err := npipe.Listen(pipeAddress) + if err != nil { + return nil, "", fmt.Errorf("listen npipe: %w", err) + } + return listener, pipeAddress, nil +} + +// listenUDS 在 Windows 平台返回不支持错误。 +func listenUDS(_ string) (net.Listener, string, error) { + return nil, "", errUnsupportedTransport +} diff --git a/internal/gateway/transport/server.go b/internal/gateway/transport/server.go new file mode 100644 index 00000000..95474675 --- /dev/null +++ b/internal/gateway/transport/server.go @@ -0,0 +1,174 @@ +package transport + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "runtime" + "strings" + "sync" +) + +// Mode 表示网关 IPC 传输类型。 +type Mode string + +const ( + // ModeAuto 表示自动按平台选择传输实现。 + ModeAuto Mode = "auto" + // ModeNPipe 表示使用 Windows Named Pipe。 + ModeNPipe Mode = "npipe" + // ModeUDS 表示使用 Unix Domain Socket。 + ModeUDS Mode = "uds" +) + +var errUnsupportedTransport = errors.New("unsupported transport on current platform") + +// Handler 负责处理单条 JSON 消息并返回 JSON 响应。 +type Handler func(ctx context.Context, payload []byte) []byte + +// Config 描述网关传输层的监听参数。 +type Config struct { + Mode Mode + Endpoint string +} + +// Server 封装网关 IPC 服务监听与连接处理逻辑。 +type Server struct { + listener net.Listener + endpoint string + handler Handler + once sync.Once +} + +// NewServer 根据配置创建并绑定 IPC 服务端。 +func NewServer(cfg Config, handler Handler) (*Server, error) { + if handler == nil { + return nil, errors.New("transport handler is nil") + } + + mode, err := normalizeMode(cfg.Mode) + if err != nil { + return nil, err + } + + listener, endpoint, err := listen(mode, cfg.Endpoint) + if err != nil { + return nil, err + } + + return &Server{ + listener: listener, + endpoint: endpoint, + handler: handler, + }, nil +} + +// Endpoint 返回当前监听的 IPC 地址。 +func (s *Server) Endpoint() string { + if s == nil { + return "" + } + return s.endpoint +} + +// Serve 启动接受循环并在上下文取消时优雅退出。 +func (s *Server) Serve(ctx context.Context) error { + if s == nil || s.listener == nil { + return errors.New("transport server is not initialized") + } + + go func() { + <-ctx.Done() + _ = s.Close() + }() + + for { + conn, err := s.listener.Accept() + if err != nil { + if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { + return nil + } + return fmt.Errorf("accept connection: %w", err) + } + go s.handleConn(ctx, conn) + } +} + +// Close 关闭监听器并释放底层资源。 +func (s *Server) Close() error { + if s == nil || s.listener == nil { + return nil + } + var closeErr error + s.once.Do(func() { + closeErr = s.listener.Close() + }) + return closeErr +} + +// handleConn 按消息粒度解码请求并写回响应。 +func (s *Server) handleConn(ctx context.Context, conn net.Conn) { + defer conn.Close() + + decoder := json.NewDecoder(conn) + encoder := json.NewEncoder(conn) + for { + var payload json.RawMessage + if err := decoder.Decode(&payload); err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { + return + } + _ = encoder.Encode(map[string]any{ + "jsonrpc": "2.0", + "error": map[string]any{ + "code": -32700, + "message": "parse error", + }, + }) + return + } + + response := s.handler(ctx, payload) + if len(response) == 0 { + continue + } + if err := encoder.Encode(json.RawMessage(response)); err != nil { + return + } + } +} + +// normalizeMode 归一化传输模式并校验输入合法性。 +func normalizeMode(mode Mode) (Mode, error) { + value := strings.TrimSpace(strings.ToLower(string(mode))) + if value == "" { + value = string(ModeAuto) + } + + switch Mode(value) { + case ModeAuto, ModeNPipe, ModeUDS: + return Mode(value), nil + default: + return "", fmt.Errorf("invalid gateway transport mode: %s", mode) + } +} + +// listen 根据传输模式创建实际监听器。 +func listen(mode Mode, endpoint string) (net.Listener, string, error) { + switch mode { + case ModeAuto: + if runtime.GOOS == "windows" { + return listenNPipe(endpoint) + } + return listenUDS(endpoint) + case ModeNPipe: + return listenNPipe(endpoint) + case ModeUDS: + return listenUDS(endpoint) + default: + return nil, "", fmt.Errorf("invalid gateway transport mode: %s", mode) + } +} diff --git a/internal/gateway/transport/server_test.go b/internal/gateway/transport/server_test.go new file mode 100644 index 00000000..fc84e018 --- /dev/null +++ b/internal/gateway/transport/server_test.go @@ -0,0 +1,96 @@ +package transport + +import ( + "context" + "encoding/json" + "net" + "path/filepath" + "runtime" + "testing" + "time" +) + +func TestNewServerInvalidMode(t *testing.T) { + _, err := NewServer(Config{Mode: "invalid"}, func(_ context.Context, _ []byte) []byte { return nil }) + if err == nil { + t.Fatal("NewServer() error = nil, want invalid mode error") + } +} + +func TestNewServerUnsupportedNPipeOnUnix(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("non-windows only") + } + + _, err := NewServer(Config{Mode: ModeNPipe}, func(_ context.Context, _ []byte) []byte { return nil }) + if err == nil { + t.Fatal("NewServer() error = nil, want unsupported transport") + } +} + +func TestServerServeAndHandle(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uds dialing is unix-only in this test") + } + + socketPath := filepath.Join(t.TempDir(), "gateway.sock") + server, err := NewServer(Config{Mode: ModeUDS, Endpoint: socketPath}, func(_ context.Context, payload []byte) []byte { + return []byte(`{"jsonrpc":"2.0","id":"1","result":"Pong"}`) + }) + if err != nil { + t.Fatalf("NewServer() error = %v", err) + } + defer func() { + _ = server.Close() + }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + serveDone := make(chan error, 1) + go func() { + serveDone <- server.Serve(ctx) + }() + + conn := waitForUnixDial(t, server.Endpoint()) + defer conn.Close() + + encoder := json.NewEncoder(conn) + decoder := json.NewDecoder(conn) + if err := encoder.Encode(map[string]any{"jsonrpc": "2.0", "id": "1", "method": "core.ping"}); err != nil { + t.Fatalf("Encode() error = %v", err) + } + + var response map[string]any + if err := decoder.Decode(&response); err != nil { + t.Fatalf("Decode() error = %v", err) + } + if got := response["result"]; got != "Pong" { + t.Fatalf("response result = %v, want Pong", got) + } + + cancel() + select { + case err := <-serveDone: + if err != nil { + t.Fatalf("Serve() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Serve() did not exit after cancel") + } +} + +func waitForUnixDial(t *testing.T, endpoint string) net.Conn { + t.Helper() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + conn, err := net.Dial("unix", endpoint) + if err == nil { + return conn + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("dial unix %s timeout", endpoint) + return nil +} diff --git a/internal/gateway/transport/uds_unix.go b/internal/gateway/transport/uds_unix.go new file mode 100644 index 00000000..2d5fda76 --- /dev/null +++ b/internal/gateway/transport/uds_unix.go @@ -0,0 +1,55 @@ +//go:build !windows + +package transport + +import ( + "errors" + "fmt" + "net" + "os" + "path/filepath" + "strings" +) + +const defaultSocketFile = "neocode-gateway.sock" + +// listenUDS 在 Unix 平台创建 UDS 监听器。 +func listenUDS(endpoint string) (net.Listener, string, error) { + socketPath := strings.TrimSpace(endpoint) + if socketPath == "" { + socketPath = filepath.Join(os.TempDir(), defaultSocketFile) + } + + if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil { + return nil, "", fmt.Errorf("create socket dir: %w", err) + } + if err := os.Remove(socketPath); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, "", fmt.Errorf("cleanup stale socket: %w", err) + } + + listener, err := net.Listen("unix", socketPath) + if err != nil { + return nil, "", fmt.Errorf("listen uds: %w", err) + } + return &udsListener{Listener: listener, socketPath: socketPath}, socketPath, nil +} + +// listenNPipe 在非 Windows 平台返回不支持错误。 +func listenNPipe(_ string) (net.Listener, string, error) { + return nil, "", errUnsupportedTransport +} + +// udsListener 在关闭时同步清理 socket 文件。 +type udsListener struct { + net.Listener + socketPath string +} + +// Close 关闭监听器并删除 socket 文件。 +func (l *udsListener) Close() error { + err := l.Listener.Close() + if removeErr := os.Remove(l.socketPath); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) && err == nil { + err = removeErr + } + return err +}