X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/2daf91497c2cc70d7156f7c0d99b1451933a818a..016d46008fc9726d2617e4e3ea098113f34c7636:/daemon.go diff --git a/daemon.go b/daemon.go index c2cf33b..5ec220e 100644 --- a/daemon.go +++ b/daemon.go @@ -65,6 +65,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { }() whatsappSetDeviceProps() + logger := daemonLogger(out, "onechat: ") whatsapp, err := openWhatsApp(ctx, dir, time.Now, daemonLogger(out, "whatsapp: "), whatsappNewClient) if err != nil { @@ -110,17 +111,19 @@ func daemon(ctx context.Context, out io.Writer) (err error) { return err } - linked := "WhatsApp is not linked" + linked := "WhatsApp not linked" if whatsapp.isLinked() { - linked = "WhatsApp is linked" + linked = "WhatsApp linked" } - logger.Printf("%s is ready: socket %s, model %s at %s effort, commands run in %s, %s, active schedules: %d", agent.config.Name, listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir, linked, schedules.Len()) + logger.Printf("ready: socket %s, model %s (%s), workdir %s, %s, %d schedules", listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir, linked, schedules.Len()) var telegram *TelegramBot + if hasBot { telegram = newTelegramBot(newTelegramAPI(telegramAPIURL, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, telegramOffsetFileName), daemonLogger(out, "telegram: ")) - logger.Print("the Telegram bot waits for messages of its owner") + + logger.Print("Telegram bot started") } return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) { @@ -154,10 +157,12 @@ func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLo } h.mutex.Lock() + h.clients[client] = true h.mutex.Unlock() h.workers.Add(2) + go h.write(client) go h.read(ctx, client, l) @@ -194,8 +199,10 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) scanner := bufio.NewScanner(client.conn) scanner.Buffer(make([]byte, 64*1024), socketRequestBytes) + for ctx.Err() == nil && scanner.Scan() { var request SocketRequest + err := json.Unmarshal(scanner.Bytes(), &request) if err == nil { err = daemonRunRequest(l, h.whatsapp, request, func(text string) { @@ -251,6 +258,7 @@ func (h *DaemonHub) remove(client *DaemonClient) { delete(h.clients, client) client.once.Do(func() { close(client.events) + _ = client.conn.Close() }) } @@ -268,9 +276,11 @@ func (h *DaemonHub) Broadcast(event Event) { // CloseAll drops every client and waits for their readers and writers. func (h *DaemonHub) CloseAll() { h.mutex.Lock() + for client := range h.clients { h.remove(client) } + h.mutex.Unlock() h.workers.Wait() @@ -340,7 +350,9 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w } var workers sync.WaitGroup + workers.Add(2) + go func() { defer workers.Done() @@ -348,29 +360,37 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w }() var whatsappErr error + go func() { defer workers.Done() whatsappErr = whatsapp.Run(ctx, l, notify) + cancel() }() var schedulesErr error + workers.Add(1) + go func() { defer workers.Done() schedulesErr = schedules.Run(ctx, l) + cancel() }() var botErr error + if bot != nil { workers.Add(1) + go func() { defer workers.Done() botErr = bot.Run(ctx, l, whatsapp) + cancel() }() } @@ -381,7 +401,9 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w } cancel() + err = errors.Join(err, listener.Close()) + hub.CloseAll() workers.Wait()