Merge pull request #12 from rinocs/fix/fixed-concurrency-on-map

fix: fixed concurrency on basic map, moved to sync.Map on telegram fe…

Author: Matrix86

feeders/telegram.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/Matrix86/driplane/data"
@@ -28,9 +29,9 @@ type Telegram struct {
 	context       context.Context
 	cancelContext context.CancelFunc
 
-	userMap    map[int64]*tg.User
-	channelMap map[int64]*tg.Channel
-	chatMap    map[int64]*tg.Chat
+	userMap    sync.Map
+	channelMap sync.Map
+	chatMap    sync.Map
 
 	api *tg.Client
 }
@@ -41,9 +42,9 @@ func NewTelegramFeeder(conf map[string]string) (Feeder, error) {
 	t := &Telegram{
 		context:       context,
 		cancelContext: cancel,
-		userMap:       make(map[int64]*tg.User),
-		channelMap:    make(map[int64]*tg.Channel),
-		chatMap:       make(map[int64]*tg.Chat),
+		userMap:       sync.Map{},
+		channelMap:    sync.Map{},
+		chatMap:       sync.Map{},
 	}
 
 	if val, ok := conf["telegram.phone_number"]; ok {
@@ -77,36 +78,45 @@ func NewTelegramFeeder(conf map[string]string) (Feeder, error) {
 
 func (t *Telegram) updateMaps(e tg.Entities) {
 	for userID, user := range e.Users {
-		t.userMap[userID] = user
+		t.userMap.Store(userID, user)
 	}
 
 	for channelID, channel := range e.Channels {
-		t.channelMap[channelID] = channel
+		t.channelMap.Store(channelID, channel)
 	}
 
 	for chatID, chat := range e.Chats {
-		t.chatMap[chatID] = chat
+		t.chatMap.Store(chatID, chat)
 	}
 }
 
 func (t *Telegram) getUserByID(id int64) (*tg.User, error) {
-	if user, ok := t.userMap[id]; ok {
+	value, ok := t.userMap.Load(id)
+	if ok {
+		user := value.(*tg.User)
 		return user, nil
 	}
+
 	return nil, fmt.Errorf("user not found")
 }
 
 func (t *Telegram) getChannelByID(id int64) (*tg.Channel, error) {
-	if channel, ok := t.channelMap[id]; ok {
+	value, ok := t.channelMap.Load(id)
+	if ok {
+		channel := value.(*tg.Channel)
 		return channel, nil
 	}
+
 	return nil, fmt.Errorf("channel not found")
 }
 
 func (t *Telegram) getChatByID(id int64) (*tg.Chat, error) {
-	if chat, ok := t.chatMap[id]; ok {
+	value, ok := t.chatMap.Load(id)
+	if ok {
+		chat := value.(*tg.Chat)
 		return chat, nil
 	}
+
 	return nil, fmt.Errorf("chat not found")
 }
 
@@ -436,9 +446,9 @@ func (t *Telegram) getDialogs() {
 	for _, chat := range dialogs.Chats {
 		switch c := chat.(type) {
 		case *tg.Chat:
-			t.chatMap[c.ID] = c
+			t.chatMap.Store(c.ID, c)
 		case *tg.Channel:
-			t.channelMap[c.ID] = c
+			t.channelMap.Store(c.ID, c)
 		}
 	}
 }
@@ -579,12 +589,26 @@ func (t *Telegram) Start() {
 					log.Info("Telegram: update recovery initialized and started, listening for events")
 					// initializing the maps and printing on debug mode
 					t.getDialogs()
-					for _, c := range t.channelMap {
+					t.channelMap.Range(func(key, value interface{}) bool {
+						c, ok := value.(*tg.Channel)
+						if !ok {
+							log.Error("Telegram: not able to get channel from the Map")
+							return true
+						}
 						log.Debug("Telegram: Channel: ID=%d AccessHash=%d Username=%s Title=%s ", c.ID, c.AccessHash, c.Username, c.Title)
-					}
-					for _, c := range t.chatMap {
+						return true
+					})
+
+					t.chatMap.Range(func(key, value interface{}) bool {
+						c, ok := value.(*tg.Chat)
+						if !ok {
+							log.Error("Telegram: not able to get chat from the Map")
+							return true
+						}
 						log.Debug("Telegram: Chat: ID=%d Title=%s", c.ID, c.Title)
-					}
+
+						return true
+					})
 				},
 			})
 		}); err != nil {