From: Anderson A. Giacomolli Date: Sat, 7 Nov 2026 16:38:35 +0000 (-0300) Subject: Name and order the daemon and socket code X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/d50c22617aafed0d3d87bf4596e933b343a10b60?ds=sidebyside Name and order the daemon and socket code Every package-level name that only the daemon uses starts with daemon: its hub of socket clients is DaemonHub, a connected client is DaemonClient, and the loop that serves the agent is daemonServe. So a client of the socket does not read as the model client or the Bot API client. What the daemon and the terminal client share is in socket.go: the socket file, the size limits, the request type SocketRequest, and the check of the socket path. The file locks that the log, the daemon, the login, and the probe share are in lock.go, with their test in lock_test.go. The declarations of these files follow the order of the go-code skill. The behavior does not change. Issue: 947bddb --- diff --git a/daemon.go b/daemon.go index cedb45e..9d8402a 100644 --- a/daemon.go +++ b/daemon.go @@ -16,41 +16,15 @@ import ( ) const ( - // agentDirName is the directory of the root agent in the data directory. - agentDirName = "agent" - - // socketFileName is the Unix socket of the daemon in the data directory. - socketFileName = "onechat.sock" - - // tokenLockFileName is the lock file of the token file in the data + // daemonAgentDirName is the directory of the root agent in the data // directory. - tokenLockFileName = "auth.lock" - - // requestBytes is the most bytes of one request line of a client. - requestBytes = 16 << 20 + daemonAgentDirName = "agent" - // clientEvents is the number of events that can wait for a slow client - // before the daemon drops it. - clientEvents = 1024 - - // socketPathBytes is the longest path of a Unix socket on Linux. - socketPathBytes = 107 - - // terminalSource is the source of the user messages from the clients of - // the socket. - terminalSource = "terminal" + // daemonClientEvents is the number of events that can wait for a slow + // client before the daemon drops it. + daemonClientEvents = 1024 ) -// errTokensInUse tells that another onechat process uses the token file. -var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon") - -// clientRequest is one line from a client: "send" with the text of a user -// message, or "interrupt". -type clientRequest struct { - Type string `json:"type"` - Text string `json:"text,omitempty"` -} - // daemon runs "onechat daemon": it owns the root agent and its files and serves // the clients on the Unix socket in the data directory until ctx ends. When // telegram.json exists, it also runs the Telegram bot. It holds the lock of the @@ -93,7 +67,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { return err } - agent, err := openAgent(filepath.Join(dir, agentDirName)) + agent, err := openAgent(filepath.Join(dir, daemonAgentDirName)) if err != nil { return err } @@ -108,7 +82,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { err = errors.Join(err, memory.Close()) }() - listener, err := listenSocket(socket) + listener, err := daemonListenSocket(socket) if err != nil { return err } @@ -127,95 +101,31 @@ func daemon(ctx context.Context, out io.Writer) (err error) { } } - return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) { + return daemonServe(ctx, listener, telegram, func(notify func(Event)) (*loop, error) { return newLoop(agent, memory, client, time.Now, notify) }) } -// serveAgent runs the loop that start makes and serves the clients of -// listener until ctx ends or the loop fails. It closes listener. When bot is -// not nil, it runs the bot too, and a failure of the bot stops the agent. -func serveAgent(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*loop, error)) error { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - - hub := &hub{ - clients: make(map[*hubClient]bool), - } - - notify := hub.Broadcast - if bot != nil { - notify = func(event Event) { - hub.Broadcast(event) - bot.Notify(event) - } - } - - l, err := start(notify) - if err != nil { - return errors.Join(err, listener.Close()) - } - - var workers sync.WaitGroup - workers.Add(1) - go func() { - defer workers.Done() - - hub.Accept(ctx, listener, l) - }() - - var botErr error - if bot != nil { - workers.Add(1) - go func() { - defer workers.Done() - - botErr = bot.Run(ctx, l) - cancel() - }() - } - - err = l.Run(ctx) - if errors.Is(err, context.Canceled) { - err = nil - } - - cancel() - err = errors.Join(err, listener.Close()) - hub.CloseAll() - workers.Wait() - - return errors.Join(err, botErr) -} - -// hub holds the connected clients and sends each event to all of them. -type hub struct { +// DaemonHub holds the connected clients and sends each event to all of them. +type DaemonHub struct { mutex sync.Mutex - clients map[*hubClient]bool + clients map[*DaemonClient]bool workers sync.WaitGroup } -// hubClient is one connected client. Its events wait in a channel for the -// writer, so a slow client does not stop the agent. -type hubClient struct { - conn net.Conn - events chan Event - once sync.Once -} - // Accept serves each client of listener until the listener closes. A new // client gets a ready event with the name of the agent, and a busy event when a // turn runs. -func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) { +func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *loop) { for { conn, err := listener.Accept() if err != nil { return } - client := &hubClient{ + client := &DaemonClient{ conn: conn, - events: make(chan Event, clientEvents), + events: make(chan Event, daemonClientEvents), } h.mutex.Lock() @@ -240,7 +150,7 @@ func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) { } // write sends the events of client as JSON lines until its channel closes. -func (h *hub) write(client *hubClient) { +func (h *DaemonHub) write(client *DaemonClient) { defer h.workers.Done() encoder := json.NewEncoder(client.conn) @@ -253,17 +163,17 @@ func (h *hub) write(client *hubClient) { } // read runs the requests of client until it closes its connection. -func (h *hub) read(ctx context.Context, client *hubClient, l *loop) { +func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *loop) { defer h.workers.Done() defer h.drop(client) scanner := bufio.NewScanner(client.conn) - scanner.Buffer(make([]byte, 64*1024), requestBytes) + scanner.Buffer(make([]byte, 64*1024), socketRequestBytes) for ctx.Err() == nil && scanner.Scan() { - var request clientRequest + var request SocketRequest err := json.Unmarshal(scanner.Bytes(), &request) if err == nil { - err = runRequest(l, request) + err = daemonRunRequest(l, request) } if err != nil { @@ -277,15 +187,23 @@ func (h *hub) read(ctx context.Context, client *hubClient, l *loop) { // Send gives event to one client, or drops the client when its events do not // fit. -func (h *hub) Send(client *hubClient, event Event) { +func (h *DaemonHub) Send(client *DaemonClient, event Event) { h.mutex.Lock() defer h.mutex.Unlock() h.send(client, event) } +// drop removes client and closes its connection. +func (h *DaemonHub) drop(client *DaemonClient) { + h.mutex.Lock() + defer h.mutex.Unlock() + + h.remove(client) +} + // send gives event to client. The caller holds the mutex. -func (h *hub) send(client *hubClient, event Event) { +func (h *DaemonHub) send(client *DaemonClient, event Event) { if !h.clients[client] { return } @@ -297,27 +215,9 @@ func (h *hub) send(client *hubClient, event Event) { } } -// Broadcast gives event to every client. -func (h *hub) Broadcast(event Event) { - h.mutex.Lock() - defer h.mutex.Unlock() - - for client := range h.clients { - h.send(client, event) - } -} - -// drop removes client and closes its connection. -func (h *hub) drop(client *hubClient) { - h.mutex.Lock() - defer h.mutex.Unlock() - - h.remove(client) -} - // remove removes client, closes its connection, and ends its writer. The // caller holds the mutex. -func (h *hub) remove(client *hubClient) { +func (h *DaemonHub) remove(client *DaemonClient) { delete(h.clients, client) client.once.Do(func() { close(client.events) @@ -325,8 +225,18 @@ func (h *hub) remove(client *hubClient) { }) } +// Broadcast gives event to every client. +func (h *DaemonHub) Broadcast(event Event) { + h.mutex.Lock() + defer h.mutex.Unlock() + + for client := range h.clients { + h.send(client, event) + } +} + // CloseAll drops every client and waits for their readers and writers. -func (h *hub) CloseAll() { +func (h *DaemonHub) CloseAll() { h.mutex.Lock() for client := range h.clients { h.remove(client) @@ -336,32 +246,18 @@ func (h *hub) CloseAll() { h.workers.Wait() } -// runRequest runs one request of a client. -func runRequest(l *loop, request clientRequest) error { - switch request.Type { - case "send": - if request.Text == "" { - return errors.New("empty message") - } - - _, err := l.Submit(terminalSource, request.Text) - - return err - case "interrupt": - if !l.Interrupt() { - return errors.New("no turn runs") - } - - return nil - default: - return fmt.Errorf("unknown request %q", request.Type) - } +// DaemonClient is one connected client. Its events wait in a channel for the +// writer, so a slow client does not stop the agent. +type DaemonClient struct { + conn net.Conn + events chan Event + once sync.Once } -// listenSocket listens on the Unix socket at path, with mode 0600. The caller -// owns the chat, so a socket file at path is left from a daemon that ended, -// and listenSocket removes it. -func listenSocket(path string) (net.Listener, error) { +// daemonListenSocket listens on the Unix socket at path, with mode 0600. The +// caller owns the chat, so a socket file at path is left from a daemon that +// ended, and daemonListenSocket removes it. +func daemonListenSocket(path string) (net.Listener, error) { err := checkSocketPath(path) if err != nil { return nil, err @@ -385,23 +281,80 @@ func listenSocket(path string) (net.Listener, error) { return listener, nil } -// checkSocketPath makes sure that path fits in the address of a Unix socket. -func checkSocketPath(path string) error { - if len(path) > socketPathBytes { - return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes) +// daemonServe runs the loop that start makes and serves the clients of listener +// until ctx ends or the loop fails. It closes listener. When bot is not nil, it +// runs the bot too, and a failure of the bot stops the agent. +func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*loop, error)) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + hub := &DaemonHub{ + clients: make(map[*DaemonClient]bool), } - return nil -} + notify := hub.Broadcast + if bot != nil { + notify = func(event Event) { + hub.Broadcast(event) + bot.Notify(event) + } + } -// lockTokens takes the lock of the token file in dir. Each process that reads -// or refreshes the tokens holds it, because a refresh in one process rotates -// the refresh token of the other and logs it out. -func lockTokens(dir string) (*os.File, error) { - lock, err := lockFile(filepath.Join(dir, tokenLockFileName)) - if errors.Is(err, errLocked) { - return nil, errTokensInUse + l, err := start(notify) + if err != nil { + return errors.Join(err, listener.Close()) } - return lock, err + var workers sync.WaitGroup + workers.Add(1) + go func() { + defer workers.Done() + + hub.Accept(ctx, listener, l) + }() + + var botErr error + if bot != nil { + workers.Add(1) + go func() { + defer workers.Done() + + botErr = bot.Run(ctx, l) + cancel() + }() + } + + err = l.Run(ctx) + if errors.Is(err, context.Canceled) { + err = nil + } + + cancel() + err = errors.Join(err, listener.Close()) + hub.CloseAll() + workers.Wait() + + return errors.Join(err, botErr) +} + +// daemonRunRequest runs one request of a client. +func daemonRunRequest(l *loop, request SocketRequest) error { + switch request.Type { + case "send": + if request.Text == "" { + return errors.New("empty message") + } + + _, err := l.Submit(terminalSource, request.Text) + + return err + case "interrupt": + if !l.Interrupt() { + return errors.New("no turn runs") + } + + return nil + default: + return fmt.Errorf("unknown request %q", request.Type) + } } diff --git a/daemon_test.go b/daemon_test.go index 3d84faa..e27c6ae 100644 --- a/daemon_test.go +++ b/daemon_test.go @@ -23,7 +23,7 @@ func TestClientsSeeTheTurnAsItHappens(t *testing.T) { modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 100, 0, 5)) } - socket := startTestDaemon(t, server) + socket := daemonStartTest(t, server) first := terminalStartTestClient(t, socket) second := terminalStartTestClient(t, socket) first.WaitFor(t, "[connected to OneChat]") @@ -54,7 +54,7 @@ func TestStopInterruptsTheTurn(t *testing.T) { <-release } - socket := startTestDaemon(t, server) + socket := daemonStartTest(t, server) client := terminalStartTestClient(t, socket) client.Type("a long task") <-arrived @@ -70,7 +70,7 @@ func TestListenSocketReplacesAnOldSocketFile(t *testing.T) { path := filepath.Join(t.TempDir(), socketFileName) require.NoError(t, os.WriteFile(path, nil, 0o600)) - listener, err := listenSocket(path) + listener, err := daemonListenSocket(path) require.NoError(t, err) info, err := os.Stat(path) @@ -81,7 +81,7 @@ func TestListenSocketReplacesAnOldSocketFile(t *testing.T) { } func TestListenSocketRefusesALongPath(t *testing.T) { - _, err := listenSocket(filepath.Join(t.TempDir(), strings.Repeat("d", 100), socketFileName)) + _, err := daemonListenSocket(filepath.Join(t.TempDir(), strings.Repeat("d", 100), socketFileName)) require.ErrorContains(t, err, "set ONECHAT_DIR to a shorter path") } @@ -97,35 +97,21 @@ func TestDaemonWritesNoFileForALongSocketPath(t *testing.T) { require.Empty(t, entries) } -func TestOneProcessHoldsTheTokens(t *testing.T) { - dir := t.TempDir() - lock, err := lockTokens(dir) - require.NoError(t, err) - - _, err = lockTokens(dir) - require.ErrorIs(t, err, errTokensInUse) - require.NoError(t, lock.Close()) - - lock, err = lockTokens(dir) - require.NoError(t, err) - require.NoError(t, lock.Close()) -} - -// startTestDaemon serves a new agent on a socket in a temporary directory and +// daemonStartTest serves a new agent on a socket in a temporary directory and // returns the socket path. The commands of the agent run in a temporary // directory. -func startTestDaemon(t *testing.T, server *ModelFake) string { +func daemonStartTest(t *testing.T, server *ModelFake) string { t.Helper() - return startTestDaemonWithBot(t, server, nil) + return daemonStartTestWithBot(t, server, nil) } -// startTestDaemonWithBot is startTestDaemon with bot, which can be nil. -func startTestDaemonWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string { +// daemonStartTestWithBot is daemonStartTest with bot, which can be nil. +func daemonStartTestWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) string { t.Helper() dir := t.TempDir() - agent, err := openAgent(filepath.Join(dir, agentDirName)) + agent, err := openAgent(filepath.Join(dir, daemonAgentDirName)) require.NoError(t, err) agent.config.Workdir = t.TempDir() @@ -134,14 +120,14 @@ func startTestDaemonWithBot(t *testing.T, server *ModelFake, bot *TelegramBot) s require.NoError(t, err) socket := filepath.Join(dir, socketFileName) - listener, err := listenSocket(socket) + listener, err := daemonListenSocket(socket) require.NoError(t, err) client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) ctx, cancel := context.WithCancel(t.Context()) done := make(chan error) go func() { - done <- serveAgent(ctx, listener, bot, func(notify func(Event)) (*loop, error) { + done <- daemonServe(ctx, listener, bot, func(notify func(Event)) (*loop, error) { return newLoop(agent, memory, client, clock.Now, notify) }) }() diff --git a/lock.go b/lock.go new file mode 100644 index 0000000..56b6fc6 --- /dev/null +++ b/lock.go @@ -0,0 +1,50 @@ +package main + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "syscall" +) + +// tokenLockFileName is the lock file of the token file in the data directory. +const tokenLockFileName = "auth.lock" + +// errTokensInUse tells that another onechat process uses the token file. +var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon") + +// errLocked tells that another process owns the chat. +var errLocked = errors.New("another process owns the chat") + +// lockTokens takes the lock of the token file in dir. Each process that reads +// or refreshes the tokens holds it, because a refresh in one process rotates +// the refresh token of the other and logs it out. +func lockTokens(dir string) (*os.File, error) { + lock, err := lockFile(filepath.Join(dir, tokenLockFileName)) + if errors.Is(err, errLocked) { + return nil, errTokensInUse + } + + return lock, err +} + +// lockFile opens the file at path and takes an exclusive lock on it. The lock +// stays until the file closes or the process ends. +func lockFile(path string) (*os.File, error) { + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, fmt.Errorf("open lock file: %w", err) + } + + err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if errors.Is(err, syscall.EWOULDBLOCK) { + err = errLocked + } + + if err != nil { + return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close()) + } + + return file, nil +} diff --git a/lock_test.go b/lock_test.go new file mode 100644 index 0000000..5308dc3 --- /dev/null +++ b/lock_test.go @@ -0,0 +1,21 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestOneProcessHoldsTheTokens(t *testing.T) { + dir := t.TempDir() + lock, err := lockTokens(dir) + require.NoError(t, err) + + _, err = lockTokens(dir) + require.ErrorIs(t, err, errTokensInUse) + require.NoError(t, lock.Close()) + + lock, err = lockTokens(dir) + require.NoError(t, err) + require.NoError(t, lock.Close()) +} diff --git a/log.go b/log.go index dfb6e70..c3d7190 100644 --- a/log.go +++ b/log.go @@ -7,14 +7,10 @@ import ( "fmt" "os" "path/filepath" - "syscall" "time" "unicode/utf8" ) -// errLocked tells that another process owns the chat. -var errLocked = errors.New("another process owns the chat") - // kinds holds the kinds that a message can have. var kinds = map[string]bool{ "user": true, @@ -250,23 +246,3 @@ func validateMessage(message Message) error { return nil } - -// lockFile opens the file at path and takes an exclusive lock on it. The lock -// stays until the file closes or the process ends. -func lockFile(path string) (*os.File, error) { - file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) - if err != nil { - return nil, fmt.Errorf("open lock file: %w", err) - } - - err = syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) - if errors.Is(err, syscall.EWOULDBLOCK) { - err = errLocked - } - - if err != nil { - return nil, errors.Join(fmt.Errorf("lock chat: %w", err), file.Close()) - } - - return file, nil -} diff --git a/socket.go b/socket.go new file mode 100644 index 0000000..a454728 --- /dev/null +++ b/socket.go @@ -0,0 +1,37 @@ +package main + +import ( + "fmt" +) + +const ( + // socketFileName is the Unix socket of the daemon in the data + // directory. + socketFileName = "onechat.sock" + + // socketRequestBytes is the most bytes of one request line of a client. + socketRequestBytes = 16 << 20 + + // socketPathBytes is the longest path of a Unix socket on Linux. + socketPathBytes = 107 + + // terminalSource is the source of the user messages from the clients of + // the socket. + terminalSource = "terminal" +) + +// checkSocketPath makes sure that path fits in the address of a Unix socket. +func checkSocketPath(path string) error { + if len(path) > socketPathBytes { + return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes) + } + + return nil +} + +// SocketRequest is one line from a client: "send" with the text of a user +// message, or "interrupt". +type SocketRequest struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` +} diff --git a/telegram_test.go b/telegram_test.go index a1a1ed2..44c9c8f 100644 --- a/telegram_test.go +++ b/telegram_test.go @@ -21,7 +21,7 @@ func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) { telegram := newTelegramFake(t) bot, _ := telegramStartTestBot(t, telegram) - socket := startTestDaemonWithBot(t, server, bot) + socket := daemonStartTestWithBot(t, server, bot) terminal := terminalStartTestClient(t, socket) terminal.WaitFor(t, "[connected to OneChat]") @@ -43,7 +43,7 @@ func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) { func TestBotIgnoresEveryoneButTheOwner(t *testing.T) { telegram := newTelegramFake(t) bot, logs := telegramStartTestBot(t, telegram) - socket := startTestDaemonWithBot(t, newModelFake(t), bot) + socket := daemonStartTestWithBot(t, newModelFake(t), bot) terminal := terminalStartTestClient(t, socket) terminal.WaitFor(t, "[connected to OneChat]") @@ -81,7 +81,7 @@ func TestBotIgnoresEveryoneButTheOwner(t *testing.T) { func TestBotAnswersThatOnlyTextIsSupported(t *testing.T) { telegram := newTelegramFake(t) bot, _ := telegramStartTestBot(t, telegram) - startTestDaemonWithBot(t, newModelFake(t), bot) + daemonStartTestWithBot(t, newModelFake(t), bot) telegram.Add(TelegramUpdate{ UpdateID: 1, @@ -106,7 +106,7 @@ func TestStopFromTelegramInterruptsTheTurn(t *testing.T) { telegram := newTelegramFake(t) bot, _ := telegramStartTestBot(t, telegram) - startTestDaemonWithBot(t, server, bot) + daemonStartTestWithBot(t, server, bot) telegram.Add(TelegramUpdate{ UpdateID: 1, @@ -149,7 +149,7 @@ func TestBotStartsAtTheStoredOffset(t *testing.T) { Message: telegramOwnerMessage("new after the restart"), }) - socket := startTestDaemonWithBot(t, server, bot) + socket := daemonStartTestWithBot(t, server, bot) terminal := terminalStartTestClient(t, socket) terminal.WaitFor(t, "[connected to OneChat]") telegramWaitForSent(t, telegram, "ok") @@ -189,7 +189,7 @@ func TestBotTriesAgainAfterFailures(t *testing.T) { } bot, logs := telegramStartTestBot(t, telegram) - startTestDaemonWithBot(t, server, bot) + daemonStartTestWithBot(t, server, bot) telegram.Add(TelegramUpdate{ UpdateID: 1, @@ -236,7 +236,7 @@ func TestRepliesGoAsRichMessages(t *testing.T) { telegram := newTelegramFake(t) bot, _ := telegramStartTestBot(t, telegram) - socket := startTestDaemonWithBot(t, server, bot) + socket := daemonStartTestWithBot(t, server, bot) terminal := terminalStartTestClient(t, socket) terminal.WaitFor(t, "[connected to OneChat]") @@ -423,7 +423,7 @@ func TestBotLogsNoTokenWhenTheNetworkFails(t *testing.T) { telegram.Close() bot, logs := telegramStartTestBot(t, telegram) - startTestDaemonWithBot(t, newModelFake(t), bot) + daemonStartTestWithBot(t, newModelFake(t), bot) require.Eventually(t, func() bool { return strings.Count(logs.String(), "no valid answer") >= 2 diff --git a/terminal.go b/terminal.go index 2906204..c785a96 100644 --- a/terminal.go +++ b/terminal.go @@ -76,10 +76,10 @@ func terminalSession(ctx context.Context, conn net.Conn, in io.Reader, out io.Wr func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error { encoder := json.NewEncoder(conn) scanner := bufio.NewScanner(in) - scanner.Buffer(make([]byte, 64*1024), requestBytes) + scanner.Buffer(make([]byte, 64*1024), socketRequestBytes) for ctx.Err() == nil && scanner.Scan() { line := scanner.Text() - request := clientRequest{ + request := SocketRequest{ Type: "send", Text: line, } @@ -88,7 +88,7 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error { case "": continue case stopCommand: - request = clientRequest{ + request = SocketRequest{ Type: "interrupt", } } @@ -106,7 +106,7 @@ func terminalSendLines(ctx context.Context, conn net.Conn, in io.Reader) error { // connection closes. func terminalShowEvents(conn net.Conn, out io.Writer) error { scanner := bufio.NewScanner(conn) - scanner.Buffer(make([]byte, 64*1024), requestBytes) + scanner.Buffer(make([]byte, 64*1024), socketRequestBytes) for scanner.Scan() { var event Event err := json.Unmarshal(scanner.Bytes(), &event) diff --git a/terminal_test.go b/terminal_test.go index 9df3e0a..d86f88a 100644 --- a/terminal_test.go +++ b/terminal_test.go @@ -13,7 +13,7 @@ import ( func TestClientEndsWhenTheDaemonEnds(t *testing.T) { dir := t.TempDir() - listener, err := listenSocket(filepath.Join(dir, socketFileName)) + listener, err := daemonListenSocket(filepath.Join(dir, socketFileName)) require.NoError(t, err) accepted := make(chan net.Conn)