narayana.im/telegabber
1
0
Fork 1
mirror of https://dev.narayana.im/narayana/telegabber.git synced 2026-10-07 16:51:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Improve thread safety of resource map

Browse source
This commit is contained in:
Bohdan Horbeshko 2026-08-09 14:16:20 -04:00
parent 4e932d1e26
commit 535423398a
2 changed files with 20 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
telegram/connect.go
View file

@ -235,9 +235,10 @@ func (c *Client) Disconnect(resource string, quit bool) bool {
c.deleteResource(resource)
}
// other resources are still active
if (len(c.resources) > 0 || c.Session.KeepOnline) && !quit {
log.Infof("Resource %v for account %v has disconnected, %v remaining", resource, c.Session.Login, len(c.resources))
log.Debugf("Resources: %#v", c.resources)
remaining := c.resourcesSnapshot()
if (len(remaining) > 0 || c.Session.KeepOnline) && !quit {
log.Infof("Resource %v for account %v has disconnected, %v remaining", resource, c.Session.Login, len(remaining))
log.Debugf("Resources: %#v", remaining)
return false
}
// already disconnected

35
telegram/utils.go
View file

@ -1928,7 +1928,7 @@ func (c *Client) SendMessageToGateway(chatId int64, message *client.Message, id
sha1 := c.GetPhotoSha1(chat.Photo.Small, chatId)
size := c.GetPhotoSize(chat.Photo.Small)
for resource := range c.resourcesRange() {
for _, resource := range c.resourcesSnapshot() {
features, ok := c.XmppClientFeatures[resource]
if ok && features != nil {
for _, feature := range *features {
@ -2388,23 +2388,20 @@ func (c *Client) deleteResource(resource string) {
}
}
func (c *Client) resourcesRange() chan string {
// resourcesSnapshot copies the currently known resources under the lock, so
// callers can safely iterate/send without holding resourcesLock for the
// whole operation (which used to block concurrent addResource/deleteResource
// for as long as a caller's loop body took, e.g. a whole batch of network
// sends).
func (c *Client) resourcesSnapshot() []string {
c.locks.resourcesLock.Lock()
defer c.locks.resourcesLock.Unlock()
resourceChan := make(chan string, 1)
go func() {
defer func() {
c.locks.resourcesLock.Unlock()
close(resourceChan)
}()
for resource := range c.resources {
resourceChan <- resource
}
}()
return resourceChan
resources := make([]string, 0, len(c.resources))
for resource := range c.resources {
resources = append(resources, resource)
}
return resources
}
// resend statuses to (to another resource, for example)
@ -2800,7 +2797,7 @@ func (c *Client) subscribeToID(id int64, chat *client.Chat, firstTime bool) {
if chat != nil {
if c.Session.MUC && c.IsGroup(chat) {
if firstTime {
for resource := range c.resourcesRange() {
for _, resource := range c.resourcesSnapshot() {
gateway.InviteToMUC(id, c.jid+"/"+resource, c.xmpp)
}
}
@ -2933,7 +2930,7 @@ func (c *Client) getFromOutbox(xmppId string) string {
func (c *Client) GetCarbonFullJids(isOutgoing bool, ignoredResource string, forceFull bool) []string {
var jids []string
if isOutgoing {
for resource := range c.resourcesRange() {
for _, resource := range c.resourcesSnapshot() {
if ignoredResource == "" || resource != ignoredResource {
jids = append(jids, c.jid+"/"+resource)
}
@ -3317,7 +3314,7 @@ func (c *Client) MigrateToMUCs() {
for _, chatID := range chatIDs {
c.unsubscribe(chatID)
for resource := range c.resourcesRange() {
for _, resource := range c.resourcesSnapshot() {
gateway.InviteToMUC(chatID, c.jid+"/"+resource, c.xmpp)
}
}