package main import ( "html/template" "log" "sync" ) type Sender interface { Send(msg GenericMessage, tpl *template.Template) error } type Sleeper interface { Sleep(ch chan<- interface{}) } type ChatId interface { GetTgIds() *[]int64 } type TgChatId struct { tgChatId int64 } func (self TgChatId) GetTgIds() *[]int64 { return &[]int64{self.tgChatId} } func Consume(events <-chan GenericMessage) { go func() { for range events { } }() } func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage { returnEvents := make(chan GenericMessage) go func() { wg.Add(1) defer wg.Done() for e := range events { returnEvents <- e err := s.Send(e, tpl) if err != nil { // TODO: handle it better panic(err) } } close(returnEvents) }() return returnEvents } func tgSenderWorker(tgEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sleeper, logger *log.Logger) <-chan GenericMessage { logger.Print("Starting") messagesToSend := make(map[ChatId][]MsgContent) waitingStarted := false timeoutEvents := make(chan interface{}) returnEvents := make(chan GenericMessage) go func() { wg.Add(1) defer wg.Done() defer close(returnEvents) loop: for { select { case ev, ok := <-tgEvents: if !ok { break loop } // Collect all messages to send them at once _, messageBuilderExists := messagesToSend[ev.ChatIds] if !messageBuilderExists { messagesToSend[ev.ChatIds] = make([]MsgContent, 0) } messagesToSend[ev.ChatIds] = append(messagesToSend[ev.ChatIds], ev.Messages...) if !waitingStarted { logger.Print("Waiting for more messages to arrive before sending...") waitingStarted = true sleeper.Sleep(timeoutEvents) } case <-timeoutEvents: logger.Print("Time's up, sending all messages we've got for now.") waitingStarted = false for chatId, msgs := range messagesToSend { returnEvents <- GenericMessage{chatId, msgs} delete(messagesToSend, chatId) } } } logger.Print("Exiting") // If anything is left to be sent, send it now for chatId, msgs := range messagesToSend { returnEvents <- GenericMessage{chatId, msgs} } }() return returnEvents }