Join filters etc. to limit numbers of goroutines that aren't needed

References: https://todo.sr.ht/~michalr/hswro-alarm-bot/7

.pre-commit-config.yaml

@@ -11,6 +11,6 @@ repos:
     rev: v0.8.3
     hooks:
     -   id: go-fmt
-        args: [ -w ]
+        args: [ -w, -s ]
     -   id: go-vet-mod
     -   id: go-test-repo-mod

filters.go

@@ -6,10 +6,24 @@ import (
 	"git.sr.ht/~michalr/go-satel"
 )
 
-func FilterByType(ev <-chan satel.Event, allowedTypes []satel.ChangeType) <-chan satel.Event {
+func isBasicEventElementOkay(basicEventElement satel.BasicEventElement, allowedTypes []satel.ChangeType, allowedIndexes []int) bool {
+	for _, allowedType := range allowedTypes {
+		if allowedType == basicEventElement.Type {
+			return true
+		}
+	}
+	for _, allowedIndex := range allowedIndexes {
+		if allowedIndex == basicEventElement.Index {
+			return true
+		}
+	}
+	return false
+}
+
+func FilterByTypeOrIndex(ev <-chan satel.Event, allowedTypes []satel.ChangeType, allowedIndexes []int) <-chan satel.Event {
 	returnChan := make(chan satel.Event)
 
-	if len(allowedTypes) == 0 {
+	if (len(allowedTypes) == 0) && (len(allowedIndexes) == 0) {
 		// no allowed types == all types are allowed
 		go func() {
 			for e := range ev {
@@ -20,39 +34,14 @@ func FilterByType(ev <-chan satel.Event, allowedTypes []satel.ChangeType) <-chan
 	} else {
 		go func() {
 			for e := range ev {
-				for _, allowedType := range allowedTypes {
-					if allowedType == e.Type {
-						returnChan <- e
-						continue
+				retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
+				for _, basicEventElement := range e.BasicEvents {
+					if isBasicEventElementOkay(basicEventElement, allowedTypes, allowedIndexes) {
+						retEv.BasicEvents = append(retEv.BasicEvents, basicEventElement)
 					}
 				}
-			}
-			close(returnChan)
-		}()
-	}
-
-	return returnChan
-}
-
-func FilterByIndex(ev <-chan satel.Event, allowedIndexes []int) <-chan satel.Event {
-	returnChan := make(chan satel.Event)
-
-	if len(allowedIndexes) == 0 {
-		// no allowed indexes == all indexes are allowed
-		go func() {
-			for e := range ev {
-				returnChan <- e
-			}
-			close(returnChan)
-		}()
-	} else {
-		go func() {
-			for e := range ev {
-				for _, allowedIndex := range allowedIndexes {
-					if allowedIndex == e.Index {
-						returnChan <- e
-						continue
-					}
+				if len(retEv.BasicEvents) != 0 {
+					returnChan <- retEv
 				}
 			}
 			close(returnChan)
@@ -67,27 +56,21 @@ func FilterByLastSeen(ev <-chan satel.Event, dataStore *DataStore, logger *log.L
 
 	go func() {
 		for e := range ev {
-			lastSeen := dataStore.GetSystemState()
-			val, ok := lastSeen[EventKey{e.Type, e.Index}]
-			if !ok || val.Value != e.Value {
-				returnChan <- e
-				dataStore.SetSystemState(EventKey{e.Type, e.Index}, EventValue{e.Value})
+			retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
+			for _, basicEventElement := range e.BasicEvents {
+				lastSeen := dataStore.GetSystemState()
+				val, ok := lastSeen[EventKey{basicEventElement.Type, basicEventElement.Index}]
+				if !ok || val.Value != basicEventElement.Value {
+					retEv.BasicEvents = append(retEv.BasicEvents, basicEventElement)
+					// TODO: flush to disk only after the loop finishes
+					dataStore.SetSystemState(EventKey{basicEventElement.Type, basicEventElement.Index}, EventValue{basicEventElement.Value})
+				}
+			}
+			if len(retEv.BasicEvents) != 0 {
+				returnChan <- retEv
 			}
 		}
-		close(returnChan)
-	}()
-
-	return returnChan
-}
-
-func CallWhenClosed(ev <-chan satel.Event, cbk func()) <-chan satel.Event {
-	returnChan := make(chan satel.Event)
-
-	go func() {
-		for e := range ev {
-			returnChan <- e
-		}
-		cbk()
+		logger.Print("Satel disconnected.")
 		close(returnChan)
 	}()
 

filters_test.go

@@ -18,25 +18,25 @@ func TestSatelEventTypeFiltering(t *testing.T) {
 
 	go func() {
 		wg.Add(1)
-		for e := range FilterByType(testEvents, []satel.ChangeType{satel.ArmedPartition, satel.PartitionFireAlarm}) {
+		for e := range FilterByTypeOrIndex(testEvents, []satel.ChangeType{satel.ArmedPartition, satel.PartitionFireAlarm}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
 		wg.Done()
 	}()
 
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.DoorOpened, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.PartitionAlarm, Index: 3, Value: true}
-	testEvents <- satel.Event{Type: satel.PartitionFireAlarm, Index: 4, Value: true}
-	testEvents <- satel.Event{Type: satel.TroublePart1, Index: 5, Value: true}
-	testEvents <- satel.Event{Type: satel.ZoneTamper, Index: 6, Value: true}
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.DoorOpened, 2, true)
+	testEvents <- makeTestSatelEvent(satel.PartitionAlarm, 3, true)
+	testEvents <- makeTestSatelEvent(satel.PartitionFireAlarm, 4, true)
+	testEvents <- makeTestSatelEvent(satel.TroublePart1, 5, true)
+	testEvents <- makeTestSatelEvent(satel.ZoneTamper, 6, true)
 
 	close(testEvents)
 	wg.Wait()
 
 	assert.Len(t, receivedEvents, 2)
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.PartitionFireAlarm, Index: 4, Value: true})
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.PartitionFireAlarm, 4, true))
 }
 
 func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T) {
@@ -46,14 +46,14 @@ func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testin
 
 	go func() {
 		wg.Add(1)
-		for e := range FilterByType(testEvents, []satel.ChangeType{}) {
+		for e := range FilterByTypeOrIndex(testEvents, []satel.ChangeType{}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
 		wg.Done()
 	}()
 
 	for index, ct := range SUPPORTED_CHANGE_TYPES {
-		testEvents <- satel.Event{Type: ct, Index: index, Value: true}
+		testEvents <- makeTestSatelEvent(ct, index, true)
 	}
 
 	close(testEvents)
@@ -61,7 +61,7 @@ func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testin
 
 	assert.Len(t, receivedEvents, len(SUPPORTED_CHANGE_TYPES))
 	for index, ct := range SUPPORTED_CHANGE_TYPES {
-		assert.Contains(t, receivedEvents, satel.Event{Type: ct, Index: index, Value: true})
+		assert.Contains(t, receivedEvents, makeTestSatelEvent(ct, index, true))
 	}
 }
 
@@ -72,25 +72,25 @@ func TestSatelIndexFiltering(t *testing.T) {
 
 	go func() {
 		wg.Add(1)
-		for e := range FilterByIndex(testEvents, []int{1, 3}) {
+		for e := range FilterByTypeOrIndex(testEvents, []satel.ChangeType{}, []int{1, 3}) {
 			receivedEvents = append(receivedEvents, e)
 		}
 		wg.Done()
 	}()
 
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.DoorOpened, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.PartitionAlarm, Index: 3, Value: true}
-	testEvents <- satel.Event{Type: satel.PartitionFireAlarm, Index: 4, Value: true}
-	testEvents <- satel.Event{Type: satel.TroublePart1, Index: 5, Value: true}
-	testEvents <- satel.Event{Type: satel.ZoneTamper, Index: 6, Value: true}
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.DoorOpened, 2, true)
+	testEvents <- makeTestSatelEvent(satel.PartitionAlarm, 3, true)
+	testEvents <- makeTestSatelEvent(satel.PartitionFireAlarm, 4, true)
+	testEvents <- makeTestSatelEvent(satel.TroublePart1, 5, true)
+	testEvents <- makeTestSatelEvent(satel.ZoneTamper, 6, true)
 
 	close(testEvents)
 	wg.Wait()
 
 	assert.Len(t, receivedEvents, 2)
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.PartitionAlarm, Index: 3, Value: true})
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.PartitionAlarm, 3, true))
 }
 
 func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T) {
@@ -101,14 +101,14 @@ func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T)
 
 	go func() {
 		wg.Add(1)
-		for e := range FilterByIndex(testEvents, []int{}) {
+		for e := range FilterByTypeOrIndex(testEvents, []satel.ChangeType{}, []int{}) {
 			receivedEvents = append(receivedEvents, e)
 		}
 		wg.Done()
 	}()
 
 	for i := 0; i < myReasonableMaxIndex; i++ {
-		testEvents <- satel.Event{Type: satel.ArmedPartition, Index: i, Value: true}
+		testEvents <- makeTestSatelEvent(satel.ArmedPartition, i, true)
 	}
 
 	close(testEvents)
@@ -116,7 +116,7 @@ func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T)
 
 	assert.Len(t, receivedEvents, myReasonableMaxIndex)
 	for i := 0; i < myReasonableMaxIndex; i++ {
-		assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: i, Value: true})
+		assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, i, true))
 	}
 }
 
@@ -140,20 +140,20 @@ func TestSatelLastSeenFiltering(t *testing.T) {
 		wg.Done()
 	}()
 
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
 
 	close(testEvents)
 	wg.Wait()
 
 	assert.Len(t, receivedEvents, 3)
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false})
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 2, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, false))
 }
 
 func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
@@ -176,20 +176,20 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
 		wg.Done()
 	}()
 
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
 
 	close(testEvents)
 	wg.Wait()
 
 	assert.Len(t, receivedEvents, 3)
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true})
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false})
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 2, true))
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, false))
 
 	testEvents = make(chan satel.Event)
 	receivedEvents = make([]satel.Event, 0)
@@ -204,15 +204,15 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
 
 	receivedEvents = make([]satel.Event, 0)
 
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: false}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	testEvents <- satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
 
 	close(testEvents)
 	wg.Wait()
 
 	assert.Len(t, receivedEvents, 1)
-	assert.Contains(t, receivedEvents, satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true})
+	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
 }

go.mod

@@ -3,7 +3,7 @@ module alarm_bot
 go 1.19
 
 require (
-	git.sr.ht/~michalr/go-satel v0.0.0-20240304210948-c93d9db530fe
+	git.sr.ht/~michalr/go-satel v0.0.0-20240305205259-053187c5d1a0
 	github.com/go-telegram-bot-api/telegram-bot-api/v5 v5.5.1
 	github.com/stretchr/testify v1.8.4
 )

go.sum

@@ -4,6 +4,8 @@ git.sr.ht/~michalr/go-satel v0.0.0-20240304192450-999413ca9994 h1:uIDVu4eIJKT4AE
 git.sr.ht/~michalr/go-satel v0.0.0-20240304192450-999413ca9994/go.mod h1:J/Bnb8xBRmuEq03dvJKLf3eCwizIhGuomUY4lVGa/6U=
 git.sr.ht/~michalr/go-satel v0.0.0-20240304210948-c93d9db530fe h1:huAwK9SgnCqgrbEn5p2nBvkoAAfl4icoGj2zXWl1ItQ=
 git.sr.ht/~michalr/go-satel v0.0.0-20240304210948-c93d9db530fe/go.mod h1:J/Bnb8xBRmuEq03dvJKLf3eCwizIhGuomUY4lVGa/6U=
+git.sr.ht/~michalr/go-satel v0.0.0-20240305205259-053187c5d1a0 h1:E5gMgMCgaZltNC9nM5s8gouzbO/X51zymT7+S4LNvvM=
+git.sr.ht/~michalr/go-satel v0.0.0-20240305205259-053187c5d1a0/go.mod h1:J/Bnb8xBRmuEq03dvJKLf3eCwizIhGuomUY4lVGa/6U=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

main.go

@@ -45,8 +45,8 @@ func (self TgSender) Send(msg GenericMessage, tpl *template.Template) error {
 	return nil
 }
 
-func sendTgMessage(tgEvents chan GenericMessage, msg MsgContent, chatId int64) {
-	tgEvents <- GenericMessage{TgChatId{chatId}, []MsgContent{msg}}
+func sendTgMessage(tgEvents chan GenericMessage, msg []satel.BasicEventElement, chatId int64) {
+	tgEvents <- GenericMessage{TgChatId{chatId}, msg}
 }
 
 type RealSleeper struct {
@@ -128,8 +128,7 @@ func getPersistenceFilePath() string {
 func main() {
 	var (
 		wg       sync.WaitGroup
-		tgEvents = make(chan GenericMessage, 100)
-		sleeper  = RealSleeper{time.Second * MessageNotMoreOftenThanSeconds}
+		tgEvents = make(chan GenericMessage, 5)
 		logger   = log.New(os.Stderr, "Main", log.Lmicroseconds)
 	)
 
@@ -150,22 +149,16 @@ func main() {
 
 	dataStore := MakeDataStore(log.New(os.Stderr, "DataStore", log.Lmicroseconds), getPersistenceFilePath())
 
-	Consume(
-		SendToTg(
-			tgSenderWorker(tgEvents, &wg, sleeper, log.New(os.Stderr, "TgSender", log.Lmicroseconds)),
-			tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tpl))
+	Consume(SendToTg(tgEvents, tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tpl))
 
 	go CloseSatelOnCtrlC(s)
 
-	for e := range FilterByIndex(FilterByType(
-		FilterByLastSeen(
-			CallWhenClosed(s.Events, func() { logger.Print("Satel disconnected.") }),
-			&dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
-		allowedTypes),
-		allowedIndexes) {
+	for e := range FilterByTypeOrIndex(
+		FilterByLastSeen(s.Events, &dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
+		allowedTypes, allowedIndexes) {
 		logger.Print("Received change from SATEL: ", e)
 		for _, chatId := range chatIds {
-			sendTgMessage(tgEvents, MsgContent{e}, chatId)
+			sendTgMessage(tgEvents, e.BasicEvents, chatId)
 		}
 	}
 

message_contents.go

@@ -14,18 +14,18 @@ type SatelNameGetter interface {
 }
 
 type MsgContent struct {
-	SatelEvent satel.Event
+	SatelEvent satel.BasicEventElement
 }
 
 type SatelMsgContent struct {
-	SatelEvent satel.Event
+	SatelEvent satel.BasicEventElement
 	s          SatelNameGetter
 	logger     *log.Logger
 }
 
 type GenericMessage struct {
 	ChatIds  ChatId
-	Messages []MsgContent
+	Messages []satel.BasicEventElement
 }
 
 type SatelMessage struct {
@@ -33,10 +33,10 @@ type SatelMessage struct {
 	Messages []SatelMsgContent
 }
 
-func getSatelMessageContent(messages []MsgContent, s SatelNameGetter, logger *log.Logger) []SatelMsgContent {
+func getSatelMessageContent(messages []satel.BasicEventElement, s SatelNameGetter, logger *log.Logger) []SatelMsgContent {
 	retval := make([]SatelMsgContent, len(messages))
 	for i, v := range messages {
-		retval[i] = SatelMsgContent{v.SatelEvent, s, logger}
+		retval[i] = SatelMsgContent{v, s, logger}
 	}
 	return retval
 }
@@ -80,6 +80,14 @@ func getEmojiWhenTrueIsBad(v bool) string {
 	}
 }
 
+func getArmedPartitionStatus(v bool) string {
+	if v {
+		return "✅ - ARMED"
+	} else {
+		return "🔴 - DISARMED"
+	}
+}
+
 func (self SatelMsgContent) FormatEvent() string {
 	switch self.SatelEvent.Type {
 	case satel.ZoneViolation:
@@ -103,7 +111,7 @@ func (self SatelMsgContent) FormatEvent() string {
 	case satel.ArmedPartitionSuppressed:
 		return fmt.Sprintf("%s: %t", self.SatelEvent.Type.String(), self.SatelEvent.Value)
 	case satel.ArmedPartition:
-		return fmt.Sprintf("%s: %s", self.SatelEvent.Type.String(), getEmojiWhenTrueIsGood(self.SatelEvent.Value))
+		return getArmedPartitionStatus(self.SatelEvent.Value)
 	case satel.PartitionArmedInMode2:
 		return fmt.Sprintf("%s: %t", self.SatelEvent.Type.String(), self.SatelEvent.Value)
 	case satel.PartitionArmedInMode3:

sender_worker.go

@@ -52,52 +52,3 @@ func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger
 
 	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
-}

sender_worker_test.go

@@ -1,76 +0,0 @@
-package main
-
-import (
-	"html/template"
-	"io"
-	"log"
-	"sync"
-	"testing"
-
-	"git.sr.ht/~michalr/go-satel"
-	"github.com/stretchr/testify/assert"
-)
-
-type MockSender struct {
-	messages []GenericMessage
-}
-
-func (self *MockSender) Send(msg GenericMessage, tpl *template.Template) error {
-	self.messages = append(self.messages, msg)
-	return nil
-}
-
-type MockSleeper struct {
-	ch        *chan<- interface{}
-	callCount int
-}
-
-func (self *MockSleeper) Sleep(ch chan<- interface{}) {
-	if self.ch == nil {
-		self.ch = &ch
-	}
-	self.callCount += 1
-}
-
-type FakeChatId struct {
-	FakeId int64
-}
-
-func (self FakeChatId) GetTgIds() *[]int64 {
-	return nil
-}
-
-var (
-	messageTest1 = satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	messageTest2 = satel.Event{Type: satel.ArmedPartition, Index: 2, Value: true}
-	messageTest3 = satel.Event{Type: satel.ArmedPartition, Index: 3, Value: true}
-	messageTest4 = satel.Event{Type: satel.ArmedPartition, Index: 4, Value: true}
-	messageTest5 = satel.Event{Type: satel.ArmedPartition, Index: 5, Value: true}
-	messageTest6 = satel.Event{Type: satel.ArmedPartition, Index: 6, Value: true}
-)
-
-func TestMessageThrottling(t *testing.T) {
-	testEvents := make(chan GenericMessage)
-	wg := sync.WaitGroup{}
-	mockSender := MockSender{make([]GenericMessage, 0)}
-	mockSleeper := MockSleeper{nil, 0}
-	Consume(SendToTg(tgSenderWorker(testEvents, &wg, &mockSleeper, log.New(io.Discard, "", log.Ltime)),
-		&mockSender, &wg, log.New(io.Discard, "", log.Ltime), nil))
-	testEvents <- GenericMessage{TgChatId{123}, []MsgContent{{messageTest1}}}
-	testEvents <- GenericMessage{TgChatId{124}, []MsgContent{{messageTest3}}}
-	testEvents <- GenericMessage{TgChatId{123}, []MsgContent{{messageTest2}}}
-	testEvents <- GenericMessage{TgChatId{124}, []MsgContent{{messageTest4}}}
-	testEvents <- GenericMessage{FakeChatId{123}, []MsgContent{{messageTest6}}}
-	assert.Equal(t, 1, mockSleeper.callCount)
-	*mockSleeper.ch <- nil
-	assert.Equal(t, 1, mockSleeper.callCount)
-	testEvents <- GenericMessage{TgChatId{123}, []MsgContent{{messageTest5}}}
-	close(testEvents)
-	wg.Wait()
-	assert.Equal(t, 2, mockSleeper.callCount)
-
-	assert.Len(t, mockSender.messages, 4)
-	assert.Contains(t, mockSender.messages, GenericMessage{TgChatId{123}, []MsgContent{{messageTest1}, {messageTest2}}})
-	assert.Contains(t, mockSender.messages, GenericMessage{TgChatId{124}, []MsgContent{{messageTest3}, {messageTest4}}})
-	assert.Contains(t, mockSender.messages, GenericMessage{FakeChatId{123}, []MsgContent{{messageTest6}}})
-}

templates_test.go

@@ -36,8 +36,8 @@ func (self MockSatelNameGetter) GetName(devType satel.DeviceType, index byte) (*
 }
 
 var (
-	tplMessageTest1 = satel.Event{Type: satel.ArmedPartition, Index: 1, Value: true}
-	tplMessageTest2 = satel.Event{Type: satel.ZoneViolation, Index: 2, Value: true}
+	tplMessageTest1 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: true}
+	tplMessageTest2 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: true}
 )
 
 func TestTelegramTemplate(t *testing.T) {
@@ -47,7 +47,7 @@ func TestTelegramTemplate(t *testing.T) {
 	tpl, err := template.New("TestTemplate").Parse(TelegramMessageTemplate)
 	assert.NoError(t, err)
 	Consume(SendToTg(testEvents, &mockSender, &wg, log.New(io.Discard, "", log.Ltime), tpl))
-	testEvents <- GenericMessage{TgChatId{123}, []MsgContent{{tplMessageTest1}, {tplMessageTest2}}}
+	testEvents <- GenericMessage{TgChatId{123}, []satel.BasicEventElement{tplMessageTest1, tplMessageTest2}}
 	close(testEvents)
 	wg.Wait()
 

test_utils.go

@@ -0,0 +1,9 @@
+package main
+
+import "git.sr.ht/~michalr/go-satel"
+
+func makeTestSatelEvent(changeType satel.ChangeType, index int, val bool) satel.Event {
+	return satel.Event{
+		BasicEvents: []satel.BasicEventElement{{Type: changeType, Index: index, Value: val}},
+	}
+}