From 052e7379d71080bb8e7edfa15f945c82e628acea Mon Sep 17 00:00:00 2001
From: Sergey Matveev <stargrave@stargrave.org>
Date: Mon, 13 Mar 2023 18:30:25 +0300
Subject: [PATCH] Simpler combined statuses update and heartbeating
---
cmd/mmc/main.go | 134 +++++++++++++++++++++++++-----------------------
cmd/start | 10 ++--
2 files changed, 73 insertions(+), 71 deletions(-)
diff --git a/cmd/mmc/main.go b/cmd/mmc/main.go
index c383cff..1596747 100644
--- a/cmd/mmc/main.go
+++ b/cmd/mmc/main.go
@@ -32,6 +32,7 @@ import (
"path"
"sort"
"strings"
+ "sync"
"syscall"
"time"
@@ -43,10 +44,9 @@ import (
const CmdFile = "/FILE "
var (
- Newwin = flag.String("newwin", "cmd/newwin", "Path to newwin command")
- DebugFifo = flag.String("debug", "", "Path to debug FIFO to be created")
- DebugFd *os.File
- UmaskCur int
+ Newwin = flag.String("newwin", "cmd/newwin", "Path to newwin command")
+ DebugFd *os.File
+ UmaskCur int
)
func rewriteIfChanged(fn string, data string) {
@@ -62,23 +62,24 @@ func main() {
entrypoint := flag.String("entrypoint", mmc.GetEntrypoint(), "Entrypoint")
notifyCmd := flag.String("notify", "cmd/notify", "Path to notification handler")
heartbeatCh := flag.String("heartbeat-ch", "town-square", "Channel for heartbeating")
- userStatusFifo := flag.String("user-status", "", "Path to FIFO for user statuses")
flag.Parse()
log.SetFlags(log.Lshortfile)
log.SetOutput(os.Stdout)
UmaskCur = syscall.Umask(0)
syscall.Umask(UmaskCur)
- var err error
- if *DebugFifo != "" {
- DebugFd, err = os.OpenFile(
- *DebugFifo, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
- )
- if err != nil {
- log.Fatalln(err)
- }
- defer DebugFd.Close()
+ os.Remove("debug")
+ err := syscall.Mkfifo("debug", 0666)
+ if err != nil {
+ log.Fatalln(err)
}
+ DebugFd, err = os.OpenFile(
+ "debug", os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
+ )
+ if err != nil {
+ log.Fatalln(err)
+ }
+ defer DebugFd.Close()
login, password := mmc.FindInNetrc(*entrypoint)
if login == "" || password == "" {
@@ -112,8 +113,6 @@ func main() {
Team := teams[0]
var updateQueue []string
- LastSent := time.Now()
-
Chans := make(map[string]*model.Channel)
time.Sleep(mmc.SleepTime)
page, resp, err := c.GetChannelsForTeamForUser(Team.Id, me.Id, false, "")
@@ -208,7 +207,6 @@ func main() {
if _, err = makePost(c, ch.Id, string(data)); err != nil {
log.Println("makePost:", err)
}
- LastSent = time.Now()
}
}(ch)
}
@@ -219,9 +217,7 @@ func main() {
}
UsersDC := make(map[string]*model.Channel, len(Users))
- userIds := make([]string, 0, len(Users))
for _, u := range Users {
- userIds = append(userIds, u.Id)
pth := path.Join("users", strings.ReplaceAll(u.Username, ".", "_"))
os.MkdirAll(pth, 0777)
rewriteIfChanged(
@@ -323,54 +319,38 @@ func main() {
if _, err = makePost(c, dc.Id, string(data)); err != nil {
log.Println("makePost:", err)
}
- LastSent = time.Now()
}
}(u)
}
UserStatus := make(map[string]string)
- updateUserStatus := func() {
- statuses, resp, err := c.GetUsersStatusesByIds(userIds)
- if err != nil {
- if DebugFd != nil {
- spew.Fdump(DebugFd, resp)
- }
+ var UserStatusM sync.RWMutex
+ go func() {
+ pth := path.Join("users", "status")
+ os.Remove(pth)
+ if err := syscall.Mkfifo(pth, 0666); err != nil {
log.Fatalln(err)
}
- if DebugFd != nil {
- spew.Fdump(DebugFd, teams)
- }
- for _, s := range statuses {
- UserStatus[Users[s.UserId].Username] = s.Status
- }
- }
- if *userStatusFifo != "" {
- updateUserStatus()
- go func() {
- for {
- time.Sleep(mmc.SleepTime)
- fd, err := os.OpenFile(
- *userStatusFifo, os.O_WRONLY|os.O_APPEND, os.FileMode(0666),
- )
- if err != nil {
- log.Println("OpenFile:", *userStatusFifo, err)
- continue
- }
- statuses := make(map[string][]string)
- for name, status := range UserStatus {
- statuses[status] = append(statuses[status], name)
- }
- for status := range statuses {
- if status == "offline" {
- continue
- }
- sort.Strings(statuses[status])
- fmt.Fprintln(fd, status+":", strings.Join(statuses[status], " "))
- }
- fd.Close()
+ for {
+ time.Sleep(mmc.SleepTime)
+ fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
+ if err != nil {
+ log.Println("OpenFile:", pth, err)
+ continue
}
- }()
- }
+ statuses := make(map[string][]string)
+ UserStatusM.RLock()
+ for name, status := range UserStatus {
+ statuses[status] = append(statuses[status], name)
+ }
+ UserStatusM.RUnlock()
+ for status := range statuses {
+ sort.Strings(statuses[status])
+ fmt.Fprintln(fd, status+":", strings.Join(statuses[status], " "))
+ }
+ fd.Close()
+ }
+ }()
log.Println("syncing", len(updateQueue)/2, "rooms")
for len(updateQueue) > 0 {
@@ -476,15 +456,17 @@ func main() {
}
go func() {
wc.Listen()
+ wc.GetStatuses()
t := time.NewTicker(time.Minute)
for {
select {
case <-t.C:
- updateUserStatus()
- if time.Now().Before(LastSent.Add(time.Minute)) {
- continue
+ if wc.ListenError != nil {
+ log.Println("ListenError:", wc.ListenError)
+ needsShutdown <- syscall.SIGTERM
+ return
}
- wc.SendMessage("ping", nil)
+ wc.GetStatuses()
if *heartbeatCh != "" {
if _, _, err = c.ViewChannel(
me.Id,
@@ -496,6 +478,7 @@ func main() {
case <-wc.PingTimeoutChannel:
log.Println("PING timeout")
needsShutdown <- syscall.SIGTERM
+ return
case e := <-wc.EventChannel:
if e == nil || !e.IsValid() {
continue
@@ -555,7 +538,9 @@ func main() {
}
case model.WebsocketEventStatusChange:
status := data["status"].(string)
+ UserStatusM.Lock()
UserStatus[user.Username] = status
+ UserStatusM.Unlock()
if *notifyCmd != "" {
exec.Command(*notifyCmd, fmt.Sprintf(
"status: %s -> %s", user.Username, status,
@@ -573,8 +558,27 @@ func main() {
if DebugFd != nil {
spew.Fdump(DebugFd, resp)
}
- if text, ok := resp.Data["text"].(string); ok && text == "pong" {
- LastSent = time.Now()
+ statuses := make(map[string]string)
+ for userId, status := range resp.Data {
+ status, ok := status.(string)
+ if !ok {
+ continue
+ }
+ user := Users[userId]
+ if user == nil {
+ continue
+ }
+ statuses[user.Username] = status
+ }
+ if len(statuses) > 0 {
+ UserStatusM.Lock()
+ for u := range UserStatus {
+ delete(UserStatus, u)
+ }
+ for u, status := range statuses {
+ UserStatus[u] = status
+ }
+ UserStatusM.Unlock()
}
}
}
diff --git a/cmd/start b/cmd/start
index 2de7122..64648f0 100755
--- a/cmd/start
+++ b/cmd/start
@@ -4,14 +4,12 @@ cmd="$(dirname "$(realpath -- "$0")")"
unset TMUX
TMUX="tmux -S tmux.sock"
$TMUX has-session -t mmc 2>/dev/null && exit
-mkdir -p users
-rm -f debug users/status
-mkfifo debug users/status
+find . -type p -delete
[ -s tmux.conf ] || sed \
-e "s#NEWWIN#$cmd/newwin#" \
-e "s#CATFILE#$cmd/catfile#" \
-e "s#DLPANE#$cmd/dlpane#" < "$cmd"/tmux.conf > tmux.conf
-$TMUX -f tmux.conf new-session -d -n ROOT -s mmc "cat debug | tai64n | tai64nlocal ; read foo"
-$TMUX split-window -h 'while : ; do cat users/status | spc -e grn,"^online:.*" -e cya,"^away:.*" ; sleep 60 ; clear ; done'
-$TMUX split-window -v "while : ; do $cmd/mmc/mmc -debug debug -user-status users/status -newwin $cmd/newwin -notify $cmd/notify | tai64n | tai64nlocal ; printf \"\\a\" ; sleep 1 ; done"
+$TMUX -f tmux.conf new-session -d -n ROOT -s mmc "while : ; do sleep 1 ; cat debug | tai64n | tai64nlocal ; done"
+$TMUX split-window -h 'while : ; do sleep 10 ; clear ; cat users/status | spc -e grn,"^online:.*" -e cya,"^away:.*" ; done'
+$TMUX split-window -v "while : ; do $cmd/mmc/mmc -newwin $cmd/newwin -notify $cmd/notify | tai64n | tai64nlocal ; printf \"\\a\" ; sleep 1 ; done"
$TMUX attach-session
--
2.51.0