73 lines
1.6 KiB
Go
73 lines
1.6 KiB
Go
package main
|
|
|
|
import (
|
|
"log"
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
type GenericMessage struct {
|
|
chatId int64
|
|
msg string
|
|
}
|
|
|
|
type Sender interface {
|
|
Send(msg GenericMessage) error
|
|
}
|
|
|
|
type Sleeper interface {
|
|
Sleep(ch chan<- interface{})
|
|
}
|
|
|
|
func tgSenderWorker(tgEvents <-chan GenericMessage, s Sender, wg *sync.WaitGroup, sleeper Sleeper, logger *log.Logger) {
|
|
logger.Print("Starting")
|
|
wg.Add(1)
|
|
defer wg.Done()
|
|
messagesToSend := make(map[int64]*strings.Builder)
|
|
waitingStarted := false
|
|
timeoutEvents := make(chan interface{})
|
|
|
|
loop:
|
|
for {
|
|
select {
|
|
case ev, ok := <-tgEvents:
|
|
if !ok {
|
|
break loop
|
|
}
|
|
// Collect all messages to send them at once
|
|
_, messageBuilderExists := messagesToSend[ev.chatId]
|
|
if !messageBuilderExists {
|
|
messagesToSend[ev.chatId] = &strings.Builder{}
|
|
}
|
|
messagesToSend[ev.chatId].WriteString(ev.msg)
|
|
messagesToSend[ev.chatId].WriteRune('\n')
|
|
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, msgBuilder := range messagesToSend {
|
|
err := s.Send(GenericMessage{chatId, msgBuilder.String()})
|
|
if err != nil {
|
|
// TODO: handle it better
|
|
panic(err)
|
|
}
|
|
delete(messagesToSend, chatId)
|
|
}
|
|
}
|
|
}
|
|
|
|
logger.Print("Exiting")
|
|
// If anything is left to be sent, send it now
|
|
for chatId, msgBuilder := range messagesToSend {
|
|
err := s.Send(GenericMessage{chatId, msgBuilder.String()})
|
|
if err != nil {
|
|
// TODO: handle it better
|
|
panic(err)
|
|
}
|
|
}
|
|
}
|