narayana/telegabber
6
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Asynchronous message processing with guaranteed sequential per-chat delivery

Browse source
This commit is contained in:
bodqhrohro 2019-12-30 07:01:56 +02:00
parent 307d5136d4
commit 7ea5e9ac73
2 changed files with 96 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
telegram/client.go
View file

@ -55,6 +55,7 @@ type Client struct {
type clientLocks struct { type clientLocks struct {
authorizationReady sync.WaitGroup authorizationReady sync.WaitGroup
chatMessageLocks map[int64]*sync.Mutex
} }
// NewClient instantiates a Telegram App // NewClient instantiates a Telegram App
@ -107,6 +108,8 @@ func NewClient(conf config.TelegramConfig, jid string, component *xmpp.Component
content: &conf.Content, content: &conf.Content,
cache: cache.NewCache(), cache: cache.NewCache(),
options: options, options: options,
locks: clientLocks{}, locks: clientLocks{
chatMessageLocks: make(map[int64]*sync.Mutex),
},
}, nil }, nil
} }

165
telegram/handlers.go
View file

@ -35,6 +35,16 @@ func int64SliceToStringSlice(ints []int64) []string {
return strings return strings
} }
func (c *Client) getChatMessageLock(chatID int64) *sync.Mutex {
lock, ok := c.locks.chatMessageLocks[chatID]
if !ok {
lock = &sync.Mutex{}
c.locks.chatMessageLocks[chatID] = lock
}
return lock
}
func (c *Client) updateHandler() { func (c *Client) updateHandler() {
listener := c.client.GetListener() listener := c.client.GetListener()
defer listener.Close() defer listener.Close()
@ -121,94 +131,103 @@ func (c *Client) updateUserStatus(update *client.UpdateUserStatus) {
// new chat discovered // new chat discovered
func (c *Client) updateNewChat(update *client.UpdateNewChat) { func (c *Client) updateNewChat(update *client.UpdateNewChat) {
if update.Chat != nil && update.Chat.Photo != nil && update.Chat.Photo.Small != nil { go func() {
_, err := c.client.DownloadFile(&client.DownloadFileRequest{ if update.Chat != nil && update.Chat.Photo != nil && update.Chat.Photo.Small != nil {
FileId: update.Chat.Photo.Small.Id, _, err := c.client.DownloadFile(&client.DownloadFileRequest{
Priority: 32, FileId: update.Chat.Photo.Small.Id,
Synchronous: true, Priority: 32,
}) Synchronous: true,
})
if err != nil { if err != nil {
log.Error("Failed to download the chat photo") log.Error("Failed to download the chat photo")
}
} }
}
c.cache.SetChat(update.Chat.Id, update.Chat) c.cache.SetChat(update.Chat.Id, update.Chat)
var isChannel = false var isChannel = false
if update.Chat.Type.ChatTypeType() == client.TypeChatTypeSupergroup { if update.Chat.Type.ChatTypeType() == client.TypeChatTypeSupergroup {
typeSupergroup, ok := update.Chat.Type.(*client.ChatTypeSupergroup) typeSupergroup, ok := update.Chat.Type.(*client.ChatTypeSupergroup)
if !ok { if !ok {
uhOh() uhOh()
}
isChannel = typeSupergroup.IsChannel
} }
isChannel = typeSupergroup.IsChannel
}
if !(isChannel && update.Chat.LastReadInboxMessageId == 0) { if !(isChannel && update.Chat.LastReadInboxMessageId == 0) {
gateway.SendPresence( gateway.SendPresence(
c.xmpp, c.xmpp,
c.jid, c.jid,
gateway.SPFrom(strconv.FormatInt(update.Chat.Id, 10)), gateway.SPFrom(strconv.FormatInt(update.Chat.Id, 10)),
gateway.SPType("subscribe"), gateway.SPType("subscribe"),
gateway.SPNickname(update.Chat.Title), gateway.SPNickname(update.Chat.Title),
) )
} }
if update.Chat.Id < 0 { if update.Chat.Id < 0 {
go c.processStatusUpdate(update.Chat.Id, update.Chat.Title, "chat") c.processStatusUpdate(update.Chat.Id, update.Chat.Title, "chat")
} }
}()
} }
// message received // message received
func (c *Client) updateNewMessage(update *client.UpdateNewMessage) { func (c *Client) updateNewMessage(update *client.UpdateNewMessage) {
// ignore self outgoing messages go func() {
if update.Message.IsOutgoing && // guarantee sequential message delivering per chat
update.Message.SendingState != nil && lock := c.getChatMessageLock(update.Message.ChatId)
update.Message.SendingState.MessageSendingStateType() == client.TypeMessageSendingStatePending { lock.Lock()
return defer lock.Unlock()
}
log.WithFields(log.Fields{ // ignore self outgoing messages
"chat_id": update.Message.ChatId, if update.Message.IsOutgoing &&
}).Warn("New message from chat") update.Message.SendingState != nil &&
update.Message.SendingState.MessageSendingStateType() == client.TypeMessageSendingStatePending {
text := c.messageToText(update.Message) return
file, filename := c.contentToFilename(update.Message.Content)
// download file(s)
if file != nil && !file.Local.IsDownloadingCompleted {
c.client.DownloadFile(&client.DownloadFileRequest{
FileId: file.Id,
Priority: 32,
Synchronous: true,
})
}
// OTR support (I do not know why would you need it, seriously)
if !strings.HasPrefix(text, "?OTR") {
var prefix strings.Builder
prefix.WriteString(c.messageToPrefix(update.Message, c.formatContent(file, filename)))
if text != "" {
// \n if it is groupchat and message is not empty
if update.Message.ChatId < 0 {
prefix.WriteString("\n")
} else if update.Message.ChatId > 0 {
prefix.WriteString(" | ")
}
prefix.WriteString(text)
} }
text = prefix.String() log.WithFields(log.Fields{
} "chat_id": update.Message.ChatId,
}).Warn("New message from chat")
// mark message as read text := c.messageToText(update.Message)
c.client.ViewMessages(&client.ViewMessagesRequest{ file, filename := c.contentToFilename(update.Message.Content)
ChatId: update.Message.ChatId,
MessageIds: []int64{update.Message.Id}, // download file(s)
ForceRead: true, if file != nil && !file.Local.IsDownloadingCompleted {
}) c.client.DownloadFile(&client.DownloadFileRequest{
// forward message to XMPP FileId: file.Id,
gateway.SendMessage(c.jid, strconv.FormatInt(update.Message.ChatId, 10), text, c.xmpp) Priority: 32,
Synchronous: true,
})
}
// OTR support (I do not know why would you need it, seriously)
if !strings.HasPrefix(text, "?OTR") {
var prefix strings.Builder
prefix.WriteString(c.messageToPrefix(update.Message, c.formatContent(file, filename)))
if text != "" {
// \n if it is groupchat and message is not empty
if update.Message.ChatId < 0 {
prefix.WriteString("\n")
} else if update.Message.ChatId > 0 {
prefix.WriteString(" | ")
}
prefix.WriteString(text)
}
text = prefix.String()
}
// mark message as read
c.client.ViewMessages(&client.ViewMessagesRequest{
ChatId: update.Message.ChatId,
MessageIds: []int64{update.Message.Id},
ForceRead: true,
})
// forward message to XMPP
gateway.SendMessage(c.jid, strconv.FormatInt(update.Message.ChatId, 10), text, c.xmpp)
}()
} }
// message content updated // message content updated