]> Sergey Matveev's repositories - mmc.git/blobdiff - cmd/mmc/main.go
Ability to send multiple /FILEs at once
[mmc.git] / cmd / mmc / main.go
index c383cff2c1c92f0ea645440989a8bb8b850316f6..f27ec75504f6ec9ed05cce84bf38a2b49920b48e 100644 (file)
@@ -32,6 +32,7 @@ import (
        "path"
        "sort"
        "strings"
+       "sync"
        "syscall"
        "time"
 
@@ -40,13 +41,10 @@ import (
        "go.stargrave.org/mmc"
 )
 
-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) {
@@ -58,27 +56,34 @@ func rewriteIfChanged(fn string, data string) {
        }
 }
 
+func mkFifo(pth string) {
+       if _, err := os.Stat(pth); err == nil {
+               return
+       }
+       if err := syscall.Mkfifo(pth, 0666); err != nil {
+               log.Fatalln(err)
+       }
+}
+
 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)
 
+       mkFifo("debug")
        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()
+       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 +117,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, "")
@@ -151,10 +154,7 @@ func main() {
                }
 
                usersPth := path.Join(pth, "users")
-               os.Remove(usersPth)
-               if err := syscall.Mkfifo(usersPth, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(usersPth)
                go func(ch *model.Channel) {
                        for {
                                time.Sleep(mmc.SleepTime)
@@ -190,10 +190,7 @@ func main() {
                }(ch)
 
                pth = path.Join(pth, "in")
-               os.Remove(pth)
-               if err := syscall.Mkfifo(pth, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(pth)
                go func(ch *model.Channel) {
                        for {
                                fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
@@ -208,7 +205,6 @@ func main() {
                                if _, err = makePost(c, ch.Id, string(data)); err != nil {
                                        log.Println("makePost:", err)
                                }
-                               LastSent = time.Now()
                        }
                }(ch)
        }
@@ -219,9 +215,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(
@@ -257,10 +251,7 @@ func main() {
                }
 
                statusPth := path.Join(pth, "status")
-               os.Remove(statusPth)
-               if err := syscall.Mkfifo(statusPth, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(statusPth)
                go func(u *model.User) {
                        for {
                                time.Sleep(mmc.SleepTime)
@@ -289,10 +280,7 @@ func main() {
                }(u)
 
                pth = path.Join(pth, "in")
-               os.Remove(pth)
-               if err := syscall.Mkfifo(pth, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(pth)
                go func(u *model.User) {
                        var dc *model.Channel
                        for {
@@ -323,54 +311,40 @@ 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")
+               mkFifo(pth)
+               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
                        }
-                       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()
+                       agg := make(map[string][]string)
+                       UserStatusM.RLock()
+                       for name, status := range UserStatus {
+                               agg[status] = append(agg[status], name)
                        }
-               }()
-       }
+                       UserStatusM.RUnlock()
+                       statuses := make([]string, 0, len(agg))
+                       for status := range agg {
+                               sort.Strings(agg[status])
+                               statuses = append(statuses, status)
+                       }
+                       sort.Strings(statuses)
+                       for _, status := range statuses {
+                               fmt.Fprintln(fd, status+":", strings.Join(agg[status], " "))
+                       }
+                       fd.Close()
+               }
+       }()
 
        log.Println("syncing", len(updateQueue)/2, "rooms")
        for len(updateQueue) > 0 {
@@ -388,15 +362,9 @@ func main() {
        go func() {
                os.MkdirAll("file", 0777)
                pthGet := path.Join("file", "get")
-               os.Remove(pthGet)
-               if err := syscall.Mkfifo(pthGet, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(pthGet)
                pthOut := path.Join("file", "out")
-               os.Remove(pthOut)
-               if err := syscall.Mkfifo(pthOut, 0666); err != nil {
-                       log.Fatalln(err)
-               }
+               mkFifo(pthOut)
                for {
                        time.Sleep(mmc.SleepTime)
                        fd, err := os.OpenFile(pthGet, os.O_RDONLY, os.FileMode(0666))
@@ -476,15 +444,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 +466,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 +526,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 +546,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()
                                }
                        }
                }