work in progress

README.md

@@ -24,6 +24,7 @@ ChirpNestBot
 - [ ] Link TelegramUserID and UserID 
 - [ ] Issue with Squire in the /g_roles ?
 - [x] Foray interception
+- [ ] Handle Foray timeout to stop spamming in case something goes bad
 - [x] Export/import all messages
 - [x] Test HTML in message
 - [x] Update auction from broadcast

client.go

@@ -7,32 +7,21 @@ import (
 	"time"
 )
 
-func clientKeepAlive(k, v interface{}) bool {
+func getLockedClient(int64 id, bool createMissing) (*ChirpClient, bool) {
-	clt := v.(*MQKeepAlive)
+	muxClients.Lock()
-	if clt.Date.Add(3 * KeepAliveHeartBeatSeconds * time.Second).Before(time.Now()) {
+	if c, ok := clients[id]; ok {
-		msgs, err := clientsQueue[clt.UserID64].Channel.QueuePurge(clientsQueue[clt.UserID64].Queue.Name, false)
+		c.Mux.Lock()
-		logOnError(err, "clientKeepAlive : Channel.QueuePurge()")
+		return c, true
-		err = clientsQueue[clt.UserID64].Channel.Close()
+	} else if createMissing {
-		logOnError(err, "clientKeepAlive : Channel.Close()")
+		c := new(ChirpClient)
-		err = clientsQueue[clt.UserID64].Connection.Close()
+		clients[id] = c
-		logOnError(err, "clientKeepAlive : Connection.Close()")
+		c.TGUserID64 = id
-		c := TGCommand{
+		c.Lock()
-			Type:       commandSendMsg,
+		return c, true
-			ToUserID64: clt.UserID64,
+	} else {
-			Text:       "Timeout, purging and closing command queue.",
+		return 0, false
-		}
-		TGCmdQueue <- c
-		c = TGCommand{
-			Type:       commandSendMsg,
-			ToUserID64: cfg.Bot.Admin,
-			Text:       fmt.Sprintf("Client %s timed out (%d messages purged).", clt.Nickname, msgs),
-		}
-		TGCmdQueue <- c
-
-		clientsKeepAlive.Delete(clt.UserID64)
-
 	}
-	return true
+	defer muxClients.Unlock()
 }
 
 func clientSendCWMsg(userID64 int64, s string) {

def.go

@@ -12,13 +12,6 @@ type DataBackup struct {
 	Messages []ChatWarsMessage `json:"messages"`
 }
 
-type MQKeepAlive struct {
-	TGUserID64 int64     `json:"tg_user_id"`
-	Nickname   string    `json:"nick"`
-	Queue      string    `json:"queue"`
-	Date       time.Time `json:"date"`
-}
-
 type MQClient struct {
 	User       string
 	Password   string
@@ -30,14 +23,25 @@ type MQClient struct {
 	Queue      amqp.Queue
 }
 
-type ChatWarsClient struct {
+type ChirpClient struct {
-	UserID64     int64     `json:"user_id"`
+	HeartBeat    time.Time `json:"heart_beat"`
-	TelegramID64 int64     `json:"telegram_id"`
+	Active       bool
-	GuildID64    int64     `json:"guild_id"`
+	TGUserID64   int64     `json:"tg_user_id"`
-	Role         string    `json:"role"`
+	MQ           MQClient  `json:"mq_client"`
-	State        string    `json:"state"`
+	CWUserID64   int64     `json:"user_id"`
-	BusyUntil    time.Time `json:"busy_until"`
+	CWGuildID64  int64     `json:"guild_id"`
-	LastUpdate   time.Time `json:"last_update"`
+	CWRole       string    `json:"role"`
+	CWState      string    `json:"state"`
+	CWBusyUntil  time.Time `json:"busy_until"`
+	CWLastUpdate time.Time `json:"last_update"`
+	Mux          sync.Mutex
+}
+
+type MQKeepAlive struct {
+	TGUserID64 int64     `json:"tg_user_id"`
+	Nickname   string    `json:"nick"`
+	Queue      string    `json:"queue"`
+	Date       time.Time `json:"date"`
 }
 
 type TGCommand struct {

main.go

@@ -60,11 +60,9 @@ var (
 	JobQueue            chan Job
 
 	msgParsingRules map[int]MessageParsingRule
-	clientsQueue    map[int64]*MQClient
 
-	clientsCW        *sync.Map
+	clients    map[int64]*ChirpClient
-	clientsKeepAlive *sync.Map
+	muxClients sync.Mutex
-	muxKeepAlive     sync.Mutex
 )
 
 func PrintText(m *tb.Message) {

workers.go

@@ -161,13 +161,13 @@ func MQKeepAliveWorker() {
 			err = json.Unmarshal(d.Body, &x)
 			logOnError(err, "MQKeepAliveWorker : Can't unmarshal.\n"+string(d.Body))
 			if err == nil {
-				muxKeepAlive.Lock()
 				if x.Date.Add(10 * time.Second).Before(time.Now()) {
 					// outdated keep-alive coming from client
-				} else if v, ok := clientsKeepAlive.Load(x.TGUserID64); ok {
+				} else if clt, ok := getLockedClient(x.TGUserID64, true); ok {
-					k := v.(*MQKeepAlive)
+					clt.HeartBeat = x.Date
-					k.Date = x.Date
+					if clt.Active {
-				} else {
+						clt.Mux.Unlock()
+					} else {
 					clt := MQClient{
 						User:     cfg.Rabbit.User,
 						Password: cfg.Rabbit.Password,
@@ -566,7 +566,31 @@ func MQTidyKeepAliveWorker() {
 	//log.Printf("MQTidyKeepAliveWorker : Starting.")
 	for true {
 		t := time.Now()
-		clientsKeepAlive.Range(clientKeepAlive)
+		muxClients.Lock()
+		for id, clt := range clients {
+			if clt.Active && clt.Date.Add(3 * KeepAliveHeartBeatSeconds * time.Second).Before(time.Now()) {
+				msgs, err := clt.MQ.Channel.QueuePurge(c.MQ.Queue.Name, false)
+				logOnError(err, "MQTidyKeepAliveWorker : Channel.QueuePurge()")
+				err = clt.MQ.Channel.Close()
+				logOnError(err, "MQTidyKeepAliveWorker : Channel.Close()")
+				err = clt.MQ.Connection.Close()
+				logOnError(err, "MQTidyKeepAliveWorker : Connection.Close()")
+				cmd := TGCommand{
+					Type:       commandSendMsg,
+					ToUserID64: clt.TGUserID64,
+					Text:       "Timeout, purging and closing command queue.",
+				}
+				TGCmdQueue <- cmd
+				cmd = TGCommand{
+					Type:       commandSendMsg,
+					ToUserID64: cfg.Bot.Admin,
+					Text:       fmt.Sprintf("Client %s timed out (%d messages purged).", clt.MQ.User, msgs),
+				}
+				TGCmdQueue <- cmd
+				clt.Active = false
+			}
+		}
+		muxClients.Unlock()
 		time.Sleep(time.Until(t.Add(time.Second)))
 	}
 	log.Printf("MQTidyKeepAliveWorker : Closing.")