Update how mattermost connector works to be more resilient to deco
This commit is contained in:
parent
1d8b7ef84a
commit
d2791094d9
1 changed files with 48 additions and 23 deletions
|
@ -42,6 +42,7 @@ type mmCaches struct {
|
||||||
mmusers map[string]string // map mm username to mm user id
|
mmusers map[string]string // map mm username to mm user id
|
||||||
sentjoined map[string]bool // map username/room name to bool
|
sentjoined map[string]bool // map username/room name to bool
|
||||||
displayname map[UserID]string // map username to last displayname
|
displayname map[UserID]string // map username to last displayname
|
||||||
|
initsynced map[string]bool // chans for which init sync has been done
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mm *Mattermost) SetHandler(h Handler) {
|
func (mm *Mattermost) SetHandler(h Handler) {
|
||||||
|
@ -57,6 +58,15 @@ func (mm *Mattermost) Configure(c Configuration) error {
|
||||||
mm.Close()
|
mm.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reinitialize shared data structures
|
||||||
|
mm.handlerStopChan = make(chan bool)
|
||||||
|
|
||||||
|
mm.caches.mmusers = make(map[string]string)
|
||||||
|
mm.caches.sentjoined = make(map[string]bool)
|
||||||
|
mm.caches.displayname = make(map[UserID]string)
|
||||||
|
mm.caches.initsynced = make(map[string]bool)
|
||||||
|
|
||||||
|
// Read config
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
mm.server, err = c.GetString("server")
|
mm.server, err = c.GetString("server")
|
||||||
|
@ -100,23 +110,23 @@ func (mm *Mattermost) Configure(c Configuration) error {
|
||||||
if token != "" {
|
if token != "" {
|
||||||
password = "token=" + token
|
password = "token=" + token
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Try to log in
|
||||||
mm.conn = matterclient.New(mm.username, password, anyteam, mm.server)
|
mm.conn = matterclient.New(mm.username, password, anyteam, mm.server)
|
||||||
mm.conn.Credentials.NoTLS = notls
|
mm.conn.Credentials.NoTLS = notls
|
||||||
err = mm.conn.Login()
|
err = mm.conn.Login()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Try to start listening for messages
|
||||||
|
// Everytime the listener reconnects, mm.handleConnected does a sync of room status
|
||||||
|
mm.conn.OnWsConnect = mm.handleConnected
|
||||||
go mm.conn.WsReceiver()
|
go mm.conn.WsReceiver()
|
||||||
go mm.conn.StatusLoop()
|
go mm.conn.StatusLoop()
|
||||||
|
go mm.handleLoop(mm.conn.MessageChan, mm.handlerStopChan)
|
||||||
|
|
||||||
for i := 0; i < 42; i++ {
|
return nil
|
||||||
time.Sleep(time.Duration(1) * time.Second)
|
|
||||||
if mm.conn.WsConnected {
|
|
||||||
mm.handleConnected()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return fmt.Errorf("Failed to connect after 42s attempting")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mm *Mattermost) User() UserID {
|
func (mm *Mattermost) User() UserID {
|
||||||
|
@ -319,17 +329,7 @@ func (mm *Mattermost) Close() {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (mm *Mattermost) handleConnected() {
|
func (mm *Mattermost) handleConnected() {
|
||||||
// Reinitialize shared data structures
|
log.Debugf("(Re-)connected to mattermost: %s@%s ; doing channel sync\n", mm.username, mm.server)
|
||||||
mm.handlerStopChan = make(chan bool)
|
|
||||||
|
|
||||||
mm.caches.mmusers = make(map[string]string)
|
|
||||||
mm.caches.sentjoined = make(map[string]bool)
|
|
||||||
mm.caches.displayname = make(map[UserID]string)
|
|
||||||
|
|
||||||
log.Debugf("Connected to mattermost: %s@%s\n", mm.username, mm.server)
|
|
||||||
|
|
||||||
// Handle incoming messages
|
|
||||||
go mm.handleLoop(mm.conn.MessageChan, mm.handlerStopChan)
|
|
||||||
|
|
||||||
// Initial channel sync
|
// Initial channel sync
|
||||||
chans := mm.conn.GetChannels()
|
chans := mm.conn.GetChannels()
|
||||||
|
@ -337,20 +337,43 @@ func (mm *Mattermost) handleConnected() {
|
||||||
for _, ch := range chans {
|
for _, ch := range chans {
|
||||||
if _, ok := doneCh[ch.Id]; !ok {
|
if _, ok := doneCh[ch.Id]; !ok {
|
||||||
doneCh[ch.Id] = true
|
doneCh[ch.Id] = true
|
||||||
go mm.initSyncChannel(ch)
|
go mm.syncChannel(ch)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (mm *Mattermost) syncChannel(ch *model.Channel) {
|
||||||
|
// The first time we see a chan, sync everything (member list, names, profile pictures)
|
||||||
|
must_initsync := func() bool {
|
||||||
|
mm.caches.Lock()
|
||||||
|
defer mm.caches.Unlock()
|
||||||
|
|
||||||
|
if x, ok := mm.caches.initsynced[ch.Id]; ok && x {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
mm.caches.initsynced[ch.Id] = true
|
||||||
|
return true
|
||||||
|
}()
|
||||||
|
|
||||||
|
if must_initsync {
|
||||||
|
mm.initSyncChannel(ch)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The following times, only sync missing messages
|
||||||
|
mm.backlogChannel(ch)
|
||||||
|
}
|
||||||
|
|
||||||
func (mm *Mattermost) initSyncChannel(ch *model.Channel) {
|
func (mm *Mattermost) initSyncChannel(ch *model.Channel) {
|
||||||
if len(strings.Split(ch.Name, "__")) == 2 {
|
if len(strings.Split(ch.Name, "__")) == 2 {
|
||||||
// DM channel
|
// DM channel
|
||||||
// Update remote user info
|
// Update remote user info
|
||||||
users := strings.Split(ch.Name, "__")
|
users := strings.Split(ch.Name, "__")
|
||||||
for _, uid := range users {
|
for _, uid := range users {
|
||||||
user := mm.conn.GetUser(uid)
|
if uid != mm.conn.User.Id {
|
||||||
if user != nil && uid != mm.conn.User.Id {
|
user := mm.conn.GetUser(uid)
|
||||||
mm.updateUserInfo(user)
|
if user != nil {
|
||||||
|
mm.updateUserInfo(user)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
|
@ -415,7 +438,9 @@ func (mm *Mattermost) initSyncChannel(ch *model.Channel) {
|
||||||
log.Warnf("Could not get channel members: %s", resp.Error.Error())
|
log.Warnf("Could not get channel members: %s", resp.Error.Error())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (mm *Mattermost) backlogChannel(ch *model.Channel) {
|
||||||
// Read backlog
|
// Read backlog
|
||||||
last_seen_post := mm.handler.CacheGet(fmt.Sprintf("last_seen_%s", ch.Id))
|
last_seen_post := mm.handler.CacheGet(fmt.Sprintf("last_seen_%s", ch.Id))
|
||||||
if last_seen_post != "" {
|
if last_seen_post != "" {
|
||||||
|
|
Loading…
Reference in a new issue