package main import ( "strings" "sync" ) type GenericMessage struct { chat_id int64 msg string } type Sender interface { Send(msg GenericMessage) error } type Sleeper interface { Sleep(ch chan<- interface{}) } func tg_sender_worker(tg_events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, sleeper Sleeper) { wg.Add(1) defer wg.Done() messagesToSend := make(map[int64]*strings.Builder) waitingStarted := false timeoutEvents := make(chan interface{}) loop: for { select { case ev, ok := <-tg_events: if !ok { break loop } // Collect all messages to send them at once _, messageBuilderExists := messagesToSend[ev.chat_id] if !messageBuilderExists { messagesToSend[ev.chat_id] = &strings.Builder{} } messagesToSend[ev.chat_id].WriteString(ev.msg) messagesToSend[ev.chat_id].WriteRune('\n') if !waitingStarted { waitingStarted = true sleeper.Sleep(timeoutEvents) } case <-timeoutEvents: waitingStarted = false for chat_id, msgBuilder := range messagesToSend { err := s.Send(GenericMessage{chat_id, msgBuilder.String()}) if err != nil { // TODO: handle it better panic(err) } delete(messagesToSend, chat_id) } } } // If anything is left to be sent, send it now for chat_id, msgBuilder := range messagesToSend { err := s.Send(GenericMessage{chat_id, msgBuilder.String()}) if err != nil { // TODO: handle it better panic(err) } } }