118 lines
2.4 KiB
Go
118 lines
2.4 KiB
Go
package main
|
|
|
|
import (
|
|
"log"
|
|
"sync"
|
|
)
|
|
|
|
type Sender interface {
|
|
Send(msg GenericMessage) 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 {
|
|
}
|
|
}()
|
|
}
|
|
|
|
type MsgContent interface {
|
|
TgString() string
|
|
}
|
|
|
|
type StringMsgContent struct {
|
|
msg string
|
|
}
|
|
|
|
func (self StringMsgContent) TgString() string { return self.msg }
|
|
|
|
type GenericMessage struct {
|
|
chatIds ChatId
|
|
msgs []MsgContent
|
|
}
|
|
|
|
func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage {
|
|
returnEvents := make(chan GenericMessage)
|
|
|
|
go func() {
|
|
wg.Add(1)
|
|
defer wg.Done()
|
|
for e := range events {
|
|
returnEvents <- e
|
|
err := s.Send(e)
|
|
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.msgs...)
|
|
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
|
|
}
|