feat(securebus): switch socket transport from JSON to FlatBuffers wire format
Replace length-prefixed JSON frames with length-prefixed FlatBuffers frames on the Unix domain socket transport. Uses the new Marshal/Unmarshal methods on ToolRequest/ToolResponse domain types for zero-copy serialization.
This commit is contained in:
parent
dc0559b617
commit
364746307f
2 changed files with 37 additions and 11 deletions
|
|
@ -3,7 +3,6 @@ package securebus
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
|
|
@ -20,7 +19,7 @@ const (
|
||||||
)
|
)
|
||||||
|
|
||||||
// SocketTransport implements Transport over a Unix domain socket using
|
// SocketTransport implements Transport over a Unix domain socket using
|
||||||
// length-prefixed JSON frames (4-byte big-endian length + JSON payload).
|
// length-prefixed FlatBuffers frames (4-byte big-endian length + FB payload).
|
||||||
// The server side listens for connections and dispatches requests to the
|
// The server side listens for connections and dispatches requests to the
|
||||||
// SecureBus; the client side connects and performs request/response exchanges.
|
// SecureBus; the client side connects and performs request/response exchanges.
|
||||||
type SocketTransport struct {
|
type SocketTransport struct {
|
||||||
|
|
@ -189,9 +188,18 @@ func (st *SocketTransport) Close() error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// writeFrame writes a length-prefixed JSON frame to w.
|
// writeFrame writes a length-prefixed FlatBuffers frame to w.
|
||||||
func writeFrame(w io.Writer, v interface{}) error {
|
func writeFrame(w io.Writer, v interface{}) error {
|
||||||
data, err := json.Marshal(v)
|
var data []byte
|
||||||
|
var err error
|
||||||
|
switch t := v.(type) {
|
||||||
|
case itr.ToolRequest:
|
||||||
|
data, err = t.Marshal()
|
||||||
|
case itr.ToolResponse:
|
||||||
|
data, err = t.Marshal()
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unsupported type for writeFrame: %T", v)
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("marshal: %w", err)
|
return fmt.Errorf("marshal: %w", err)
|
||||||
}
|
}
|
||||||
|
|
@ -210,7 +218,7 @@ func writeFrame(w io.Writer, v interface{}) error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// readFrame reads a length-prefixed JSON frame from r into v.
|
// readFrame reads a length-prefixed FlatBuffers frame from r into v.
|
||||||
func readFrame(r io.Reader, v interface{}) error {
|
func readFrame(r io.Reader, v interface{}) error {
|
||||||
var header [4]byte
|
var header [4]byte
|
||||||
if _, err := io.ReadFull(r, header[:]); err != nil {
|
if _, err := io.ReadFull(r, header[:]); err != nil {
|
||||||
|
|
@ -227,5 +235,21 @@ func readFrame(r io.Reader, v interface{}) error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return json.Unmarshal(buf, v)
|
switch t := v.(type) {
|
||||||
|
case *itr.ToolRequest:
|
||||||
|
req, err := itr.UnmarshalRequest(buf)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("unmarshal request: %w", err)
|
||||||
|
}
|
||||||
|
*t = req
|
||||||
|
case *itr.ToolResponse:
|
||||||
|
resp, err := itr.UnmarshalResponse(buf)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("unmarshal response: %w", err)
|
||||||
|
}
|
||||||
|
*t = resp
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("unsupported type for readFrame: %T", v)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -75,11 +75,13 @@ func TestSocketTransportMultipleRequests(t *testing.T) {
|
||||||
defer client.Close()
|
defer client.Close()
|
||||||
|
|
||||||
for i := 0; i < 10; i++ {
|
for i := 0; i < 10; i++ {
|
||||||
req := itr.ToolRequest{
|
req := itr.NewToolExecRequest(
|
||||||
ID: "req-" + string(rune('A'+i)),
|
"req-"+string(rune('A'+i)),
|
||||||
Type: itr.CmdToolExec,
|
"sess",
|
||||||
Timestamp: time.Now().UnixNano(),
|
"",
|
||||||
}
|
"echo",
|
||||||
|
`{}`,
|
||||||
|
)
|
||||||
resp, err := client.Send(context.Background(), req)
|
resp, err := client.Send(context.Background(), req)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, req.ID, resp.ID)
|
assert.Equal(t, req.ID, resp.ID)
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue