Probably last async stuff removed that's not satel

filters.go

@@ -7,84 +7,48 @@ import (
 	"git.sr.ht/~michalr/go-satel"
 )
 
-func isBasicEventElementOkay(basicEventElement satel.BasicEventElement, allowedTypes []SatelChangeType, allowedIndexes []int) bool {
+type SyncFilter[MsgType any] interface {
-	for _, allowedType := range allowedTypes {
+	Then(what SyncFilter[MsgType]) SyncFilter[MsgType]
-		if allowedType.GetChangeType() == basicEventElement.Type {
+	Call(msg MsgType)
-			return true
-		}
-	}
-	for _, allowedIndex := range allowedIndexes {
-		if allowedIndex == basicEventElement.Index {
-			return true
-		}
-	}
-	return false
 }
 
-func FilterByTypeOrIndex(ev <-chan satel.Event, wg *sync.WaitGroup, allowedTypes []SatelChangeType, allowedIndexes []int) <-chan satel.Event {
+type SyncFilterImpl[MsgType any] struct {
-	returnChan := make(chan satel.Event)
+	next SyncFilter[MsgType]
-
-	if (len(allowedTypes) == 0) && (len(allowedIndexes) == 0) {
-		// no allowed types == all types are allowed
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			defer close(returnChan)
-
-			for e := range ev {
-				returnChan <- e
-			}
-		}()
-	} else {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			defer close(returnChan)
-
-			for e := range ev {
-				retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
-				for _, basicEventElement := range e.BasicEvents {
-					if isBasicEventElementOkay(basicEventElement, allowedTypes, allowedIndexes) {
-						retEv.BasicEvents = append(retEv.BasicEvents, basicEventElement)
-					}
-				}
-				if len(retEv.BasicEvents) != 0 {
-					returnChan <- retEv
-				}
-			}
-		}()
-	}
-
-	return returnChan
 }
 
-func FilterByLastSeen(ev <-chan satel.Event, wg *sync.WaitGroup, dataStore *DataStore, logger *log.Logger) <-chan satel.Event {
+func (impl *SyncFilterImpl[MsgType]) Then(what SyncFilter[MsgType]) SyncFilter[MsgType] {
-	returnChan := make(chan satel.Event)
+	impl.next = what
+	return what
+}
 
+func (impl *SyncFilterImpl[MsgType]) CallNext(msg MsgType) {
+	if impl.next != nil {
+		impl.next.Call(msg)
+	}
+}
+
+type CollectFromChannel[MsgType any] struct{ SyncFilterImpl[MsgType] }
+
+func (collect *CollectFromChannel[MsgType]) Call(msg MsgType) {
+	collect.CallNext(msg)
+}
+
+func (collect CollectFromChannel[MsgType]) Collect(events <-chan MsgType, wg *sync.WaitGroup, onClose func()) {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		defer close(returnChan)
+		defer onClose()
 
-		for e := range ev {
+		for e := range events {
-			retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
+			collect.Call(e)
-			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
-			}
 		}
-		logger.Print("Satel disconnected.")
 	}()
+}
 
-	return returnChan
+type ThrottleSync struct {
+	SyncFilterImpl[GenericMessage]
+
+	events chan GenericMessage
 }
 
 func appendToGenericMessage(msg *GenericMessage, new *GenericMessage) *GenericMessage {
@@ -107,20 +71,19 @@ throughNewMessages:
 	return msg
 }
 
-func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sleeper, logger *log.Logger) <-chan GenericMessage {
+func MakeThrottleSync(sleeper Sleeper, logger *log.Logger, wg *sync.WaitGroup) *ThrottleSync {
-	returnChan := make(chan GenericMessage)
+	events := make(chan GenericMessage)
-	timeoutEvents := make(chan interface{})
+
+	throttle := ThrottleSync{SyncFilterImpl[GenericMessage]{}, events}
 
 	wg.Add(1)
 	go func() {
-		defer wg.Done()
+		timeoutEvents := make(chan interface{})
-		defer close(returnChan)
-
 		var currentEvent *GenericMessage = nil
 	loop:
 		for {
 			select {
-			case ev, ok := <-inputEvents:
+			case ev, ok := <-events:
 				if !ok {
 					break loop
 				}
@@ -131,16 +94,117 @@ func Throttle(inputEvents <-chan GenericMessage, wg *sync.WaitGroup, sleeper Sle
 				currentEvent = appendToGenericMessage(currentEvent, &ev)
 			case <-timeoutEvents:
 				logger.Print("Time's up, sending all messages we've got for now.")
-				returnChan <- *currentEvent
+				throttle.CallNext(*currentEvent)
 				currentEvent = nil
 			}
 		}
 
 		// If anything is left to be sent, send it now
 		if currentEvent != nil {
-			returnChan <- *currentEvent
+			throttle.CallNext(*currentEvent)
 		}
+		wg.Done()
+		logger.Print("Throttling goroutine finishing")
 	}()
 
-	return returnChan
+	return &throttle
+}
+
+func (throttle *ThrottleSync) Close() { close(throttle.events) }
+
+func (throttle *ThrottleSync) Call(msg GenericMessage) {
+	throttle.events <- msg
+}
+
+type Convert[InMsgType any] struct {
+	out     SyncFilter[GenericMessage]
+	convert func(InMsgType) GenericMessage
+}
+
+func MakeConvert[InMsgType any](convertFunc func(InMsgType) GenericMessage) *Convert[InMsgType] {
+	return &Convert[InMsgType]{nil, convertFunc}
+}
+
+func (convert *Convert[InMsgType]) Call(msg InMsgType) {
+	if convert.out == nil {
+		panic("Use ConvertTo() to set next element in the chain.")
+	}
+	convert.out.Call(convert.convert(msg))
+}
+
+func (convert *Convert[InMsgType]) ConvertTo(out SyncFilter[GenericMessage]) SyncFilter[GenericMessage] {
+	convert.out = out
+	return out
+}
+
+func (convert *Convert[InMsgType]) Then(_ SyncFilter[InMsgType]) SyncFilter[InMsgType] {
+	panic("Use ConvertTo() with Convert object instead of Then().")
+}
+
+type FilterByLastSeen struct {
+	SyncFilterImpl[satel.Event]
+	dataStore *DataStore
+}
+
+func MakeFilterByLastSeen(dataStore *DataStore) *FilterByLastSeen {
+	return &FilterByLastSeen{SyncFilterImpl[satel.Event]{}, dataStore}
+}
+
+func (filter *FilterByLastSeen) Call(ev satel.Event) {
+	retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
+	for _, basicEventElement := range ev.BasicEvents {
+		lastSeen := filter.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
+			filter.dataStore.SetSystemState(EventKey{basicEventElement.Type, basicEventElement.Index},
+				EventValue{basicEventElement.Value})
+		}
+	}
+	if len(retEv.BasicEvents) != 0 {
+		filter.CallNext(retEv)
+	}
+
+}
+
+type FilterByTypeOrIndex struct {
+	SyncFilterImpl[satel.Event]
+	allowedTypes   []SatelChangeType
+	allowedIndexes []int
+}
+
+func MakeFilterByTypeOrIndex(allowedTypes []SatelChangeType, allowedIndexes []int) *FilterByTypeOrIndex {
+	return &FilterByTypeOrIndex{SyncFilterImpl[satel.Event]{}, allowedTypes, allowedIndexes}
+}
+
+func isBasicEventElementOkay(basicEventElement satel.BasicEventElement, allowedTypes []SatelChangeType, allowedIndexes []int) bool {
+	for _, allowedType := range allowedTypes {
+		if allowedType.GetChangeType() == basicEventElement.Type {
+			return true
+		}
+	}
+	for _, allowedIndex := range allowedIndexes {
+		if allowedIndex == basicEventElement.Index {
+			return true
+		}
+	}
+	return false
+}
+
+func (filter *FilterByTypeOrIndex) Call(ev satel.Event) {
+	if (len(filter.allowedTypes) == 0) && (len(filter.allowedIndexes) == 0) {
+		// no allowed types == all types are allowed
+		filter.CallNext(ev)
+	} else {
+		retEv := satel.Event{BasicEvents: make([]satel.BasicEventElement, 0)}
+		for _, basicEventElement := range ev.BasicEvents {
+			if isBasicEventElementOkay(basicEventElement, filter.allowedTypes, filter.allowedIndexes) {
+				retEv.BasicEvents = append(retEv.BasicEvents, basicEventElement)
+			}
+		}
+		if len(retEv.BasicEvents) != 0 {
+			filter.CallNext(retEv)
+		}
+	}
 }

filters_test.go

@@ -12,115 +12,74 @@ import (
 )
 
 func TestSatelEventTypeFiltering(t *testing.T) {
-	testEvents := make(chan satel.Event)
+	tested := MakeFilterByTypeOrIndex([]SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}},
-	receivedEvents := make([]satel.Event, 0)
+		[]int{})
-	wg := sync.WaitGroup{}
+	mock := &GenericSyncMockFilter[satel.Event]{}
 
-	wg.Add(1)
+	tested.Then(mock)
-	go func() {
-		defer wg.Done()
 
-		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}}, []int{}) {
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-			receivedEvents = append(receivedEvents, e)
+	tested.Call(makeTestSatelEvent(satel.DoorOpened, 2, true))
-		}
+	tested.Call(makeTestSatelEvent(satel.PartitionAlarm, 3, true))
-	}()
+	tested.Call(makeTestSatelEvent(satel.PartitionFireAlarm, 4, true))
+	tested.Call(makeTestSatelEvent(satel.TroublePart1, 5, true))
+	tested.Call(makeTestSatelEvent(satel.ZoneTamper, 6, true))
 
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	assert.Len(t, mock.collected, 2)
-	testEvents <- makeTestSatelEvent(satel.DoorOpened, 2, true)
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.PartitionAlarm, 3, true)
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.PartitionFireAlarm, 4, 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, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.PartitionFireAlarm, 4, true))
 }
 
 func TestSatelEventTypeFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T) {
-	testEvents := make(chan satel.Event)
+	tested := MakeFilterByTypeOrIndex([]SatelChangeType{},
-	receivedEvents := make([]satel.Event, 0)
+		[]int{})
-	wg := sync.WaitGroup{}
+	mock := &GenericSyncMockFilter[satel.Event]{}
 
-	wg.Add(1)
+	tested.Then(mock)
-	go func() {
-		defer wg.Done()
-
-		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
 
 	for index, ct := range SUPPORTED_CHANGE_TYPES {
-		testEvents <- makeTestSatelEvent(ct, index, true)
+		tested.Call(makeTestSatelEvent(ct, index, true))
 	}
 
-	close(testEvents)
+	assert.Len(t, mock.collected, len(SUPPORTED_CHANGE_TYPES))
-	wg.Wait()
-
-	assert.Len(t, receivedEvents, len(SUPPORTED_CHANGE_TYPES))
 	for index, ct := range SUPPORTED_CHANGE_TYPES {
-		assert.Contains(t, receivedEvents, makeTestSatelEvent(ct, index, true))
+		assert.Contains(t, mock.collected, makeTestSatelEvent(ct, index, true))
 	}
 }
 
 func TestSatelIndexFiltering(t *testing.T) {
-	testEvents := make(chan satel.Event)
+	tested := MakeFilterByTypeOrIndex([]SatelChangeType{}, []int{1, 3})
-	receivedEvents := make([]satel.Event, 0)
+	mock := &GenericSyncMockFilter[satel.Event]{}
-	wg := sync.WaitGroup{}
 
-	wg.Add(1)
+	tested.Then(mock)
-	go func() {
-		defer wg.Done()
 
-		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{1, 3}) {
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-			receivedEvents = append(receivedEvents, e)
+	tested.Call(makeTestSatelEvent(satel.DoorOpened, 2, true))
-		}
+	tested.Call(makeTestSatelEvent(satel.PartitionAlarm, 3, true))
-	}()
+	tested.Call(makeTestSatelEvent(satel.PartitionFireAlarm, 4, true))
+	tested.Call(makeTestSatelEvent(satel.TroublePart1, 5, true))
+	tested.Call(makeTestSatelEvent(satel.ZoneTamper, 6, true))
 
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	assert.Len(t, mock.collected, 2)
-	testEvents <- makeTestSatelEvent(satel.DoorOpened, 2, true)
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.PartitionAlarm, 3, true)
+	assert.Contains(t, mock.collected, 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, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.PartitionAlarm, 3, true))
 }
 
 func TestSatelIndexFiltering_NoAllowedEventTypesMeansAllAreAllowed(t *testing.T) {
-	testEvents := make(chan satel.Event)
+	tested := MakeFilterByTypeOrIndex([]SatelChangeType{{satel.ArmedPartition}, {satel.PartitionFireAlarm}},
-	receivedEvents := make([]satel.Event, 0)
+		[]int{})
-	wg := sync.WaitGroup{}
+	mock := &GenericSyncMockFilter[satel.Event]{}
 	myReasonableMaxIndex := 100 // I wanted to use math.MaxInt at first, but it's kind of a waste of time here
 
-	wg.Add(1)
+	tested.Then(mock)
-	go func() {
-		defer wg.Done()
-
-		for e := range FilterByTypeOrIndex(testEvents, &wg, []SatelChangeType{}, []int{}) {
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
 
 	for i := 0; i < myReasonableMaxIndex; i++ {
-		testEvents <- makeTestSatelEvent(satel.ArmedPartition, i, true)
+		tested.Call(makeTestSatelEvent(satel.ArmedPartition, i, true))
 	}
 
-	close(testEvents)
+	assert.Len(t, mock.collected, myReasonableMaxIndex)
-	wg.Wait()
-
-	assert.Len(t, receivedEvents, myReasonableMaxIndex)
 	for i := 0; i < myReasonableMaxIndex; i++ {
-		assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, i, true))
+		assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, i, true))
 	}
 }
 
@@ -130,35 +89,26 @@ func TestSatelLastSeenFiltering(t *testing.T) {
 	tempFileName := f.Name()
 	assert.NoError(t, f.Close())
 	defer os.Remove(f.Name())
-	testEvents := make(chan satel.Event)
+
-	receivedEvents := make([]satel.Event, 0)
-	wg := sync.WaitGroup{}
 	fakeLog := log.New(io.Discard, "", log.Ltime)
 	ds := MakeDataStore(fakeLog, tempFileName)
 
-	wg.Add(1)
+	tested := MakeFilterByLastSeen(&ds)
-	go func() {
+	mock := &GenericSyncMockFilter[satel.Event]{}
-		defer wg.Done()
 
-		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
+	tested.Then(mock)
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
 
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
 
-	close(testEvents)
+	assert.Len(t, mock.collected, 3)
-	wg.Wait()
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	assert.Len(t, receivedEvents, 3)
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, 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) {
@@ -167,61 +117,45 @@ func TestSatelLastSeenFilteringWithPersistence(t *testing.T) {
 	tempFileName := f.Name()
 	assert.NoError(t, f.Close())
 	defer os.Remove(f.Name())
-	testEvents := make(chan satel.Event)
+
-	receivedEvents := make([]satel.Event, 0)
-	wg := sync.WaitGroup{}
 	fakeLog := log.New(io.Discard, "", log.Ltime)
 	ds := MakeDataStore(fakeLog, tempFileName)
 
-	wg.Add(1)
+	tested := MakeFilterByLastSeen(&ds)
-	go func() {
+	mock := &GenericSyncMockFilter[satel.Event]{}
-		defer wg.Done()
 
-		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
+	tested.Then(mock)
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
 
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
 
-	close(testEvents)
+	assert.Len(t, mock.collected, 3)
-	wg.Wait()
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 2, true))
+	assert.Contains(t, mock.collected, makeTestSatelEvent(satel.ArmedPartition, 1, false))
 
-	assert.Len(t, receivedEvents, 3)
+	tested = nil
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
+	mock = nil
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, false))
 
-	testEvents = make(chan satel.Event)
+	ds2 := MakeDataStore(fakeLog, tempFileName)
-	receivedEvents = make([]satel.Event, 0)
-	ds = MakeDataStore(fakeLog, tempFileName)
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
 
-		for e := range FilterByLastSeen(testEvents, &wg, &ds, fakeLog) {
+	tested2 := MakeFilterByLastSeen(&ds2)
-			receivedEvents = append(receivedEvents, e)
+	mock2 := &GenericSyncMockFilter[satel.Event]{}
-		}
-	}()
 
-	receivedEvents = make([]satel.Event, 0)
+	tested2.Then(mock2)
 
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested2.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, false)
+	tested2.Call(makeTestSatelEvent(satel.ArmedPartition, 1, false))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 1, true)
+	tested2.Call(makeTestSatelEvent(satel.ArmedPartition, 1, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested2.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
-	testEvents <- makeTestSatelEvent(satel.ArmedPartition, 2, true)
+	tested2.Call(makeTestSatelEvent(satel.ArmedPartition, 2, true))
 
-	close(testEvents)
+	assert.Len(t, mock2.collected, 1)
-	wg.Wait()
+	assert.Contains(t, mock2.collected, makeTestSatelEvent(satel.ArmedPartition, 1, true))
-
-	assert.Len(t, receivedEvents, 1)
-	assert.Contains(t, receivedEvents, makeTestSatelEvent(satel.ArmedPartition, 1, true))
 }
 
 type MockSleeper struct {
@@ -236,9 +170,59 @@ func (self *MockSleeper) Sleep(ch chan<- interface{}) {
 	self.callCount += 1
 }
 
-func TestThrottle(t *testing.T) {
+type GenericSyncMockFilter[T any] struct {
+	SyncFilterImpl[T]
+	collected []T
+}
+
+func (self *GenericSyncMockFilter[T]) Call(msg T) {
+	self.collected = append(self.collected, msg)
+	self.CallNext(msg)
+}
+
+type SyncMockFilter = GenericSyncMockFilter[GenericMessage]
+
+func TestSyncCollect(t *testing.T) {
 	testEvents := make(chan GenericMessage)
-	receivedEvents := make([]GenericMessage, 0)
+	wg := sync.WaitGroup{}
+
+	tested := CollectFromChannel[GenericMessage]{}
+	mock := &SyncMockFilter{}
+	mock2 := &SyncMockFilter{}
+
+	tested.Then(mock).Then(mock2)
+
+	wg.Add(1)
+	tested.Collect(testEvents, &wg, func() { wg.Done() })
+
+	testEvents <- makeGenericMessage(satel.ArmedPartition, 1, true)
+	testEvents <- makeGenericMessage(satel.DoorOpened, 2, true)
+	testEvents <- makeGenericMessage(satel.PartitionAlarm, 3, true)
+	testEvents <- makeGenericMessage(satel.PartitionFireAlarm, 4, true)
+	testEvents <- makeGenericMessage(satel.TroublePart1, 5, true)
+	testEvents <- makeGenericMessage(satel.ZoneTamper, 6, true)
+
+	close(testEvents)
+	wg.Wait()
+
+	assert.Len(t, mock.collected, 6)
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.ArmedPartition, 1, true))
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.DoorOpened, 2, true))
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.PartitionAlarm, 3, true))
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.PartitionFireAlarm, 4, true))
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.TroublePart1, 5, true))
+	assert.Contains(t, mock.collected, makeGenericMessage(satel.ZoneTamper, 6, true))
+
+	assert.Len(t, mock2.collected, 6)
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.ArmedPartition, 1, true))
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.DoorOpened, 2, true))
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.PartitionAlarm, 3, true))
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.PartitionFireAlarm, 4, true))
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.TroublePart1, 5, true))
+	assert.Contains(t, mock2.collected, makeGenericMessage(satel.ZoneTamper, 6, true))
+}
+
+func TestThrottleSync(t *testing.T) {
 	wg := sync.WaitGroup{}
 	fakeLog := log.New(io.Discard, "", log.Ltime)
 	mockSleeper := MockSleeper{nil, 0}
@@ -250,83 +234,57 @@ func TestThrottle(t *testing.T) {
 		tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
 	)
 
-	wg.Add(1)
+	tested := MakeThrottleSync(&mockSleeper, fakeLog, &wg)
-	go func() {
+	mock := &SyncMockFilter{}
-		defer wg.Done()
 
-		for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
+	tested.Then(mock)
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
 
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1}}
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest1}})
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest2}}
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest2}})
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest3}}
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest3}})
 	*mockSleeper.ch <- nil
 
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest4}}
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest4}})
 
-	close(testEvents)
+	tested.Close()
 	wg.Wait()
 
 	assert.Equal(t, 2, mockSleeper.callCount)
-	assert.Len(t, receivedEvents, 2)
+	assert.Len(t, mock.collected, 2)
-	assert.Contains(t, receivedEvents[0].Messages, tplMessageTest2)
+	assert.Contains(t, mock.collected[0].Messages, tplMessageTest2)
-	assert.Contains(t, receivedEvents[0].Messages, tplMessageTest3)
+	assert.Contains(t, mock.collected[0].Messages, tplMessageTest3)
-	assert.Len(t, receivedEvents[0].Messages, 2)
+	assert.Len(t, mock.collected[0].Messages, 2)
 
-	assert.Contains(t, receivedEvents[1].Messages, tplMessageTest4)
+	assert.Contains(t, mock.collected[1].Messages, tplMessageTest4)
-	assert.Len(t, receivedEvents[1].Messages, 1)
+	assert.Len(t, mock.collected[1].Messages, 1)
 }
 
-func makeMassiveEvent(element satel.BasicEventElement, numElements int) GenericMessage {
+func TestConvert_failsWhenNotConverting(t *testing.T) {
-	retval := GenericMessage{make([]satel.BasicEventElement, 0)}
+	a := assert.New(t)
+	tested := MakeConvert(func(in int) GenericMessage {
+		a.Equal(in, 1)
+		return GenericMessage{}
+	})
+	mock := &GenericSyncMockFilter[int]{}
 
-	for i := 0; i < numElements; i++ {
+	a.Panics(func() {
-		retval.Messages = append(retval.Messages, element)
+		tested.Then(mock)
-	}
+		tested.Call(1)
-
+	})
-	return retval
 }
 
-func TestThrottle_ManyMessagesInOneEvent(t *testing.T) {
+func TestConvert(t *testing.T) {
-	testEvents := make(chan GenericMessage)
+	a := assert.New(t)
-	receivedEvents := make([]GenericMessage, 0)
+	numCalled := 0
-	wg := sync.WaitGroup{}
+	tested := MakeConvert(func(in int) GenericMessage {
-	fakeLog := log.New(io.Discard, "", log.Ltime)
+		a.Equal(in, 1)
-	mockSleeper := MockSleeper{nil, 0}
+		numCalled += 1
+		return GenericMessage{}
+	})
+	mock := &SyncMockFilter{}
 
-	var (
+	tested.ConvertTo(mock)
-		tplMessageTest1 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: true}
+	tested.Call(1)
-		tplMessageTest2 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: true}
-		tplMessageTest3 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: false}
-		tplMessageTest4 = satel.BasicEventElement{Type: satel.ZoneViolation, Index: 2, Value: false}
-	)
 
-	wg.Add(1)
+	a.Equal(numCalled, 1)
-	go func() {
-		defer wg.Done()
-		for e := range Throttle(testEvents, &wg, &mockSleeper, fakeLog) {
-			receivedEvents = append(receivedEvents, e)
-		}
-	}()
-
-	testEvents <- makeMassiveEvent(tplMessageTest1, 100)
-	testEvents <- makeMassiveEvent(tplMessageTest2, 100)
-	testEvents <- makeMassiveEvent(tplMessageTest3, 100)
-	*mockSleeper.ch <- nil
-
-	testEvents <- makeMassiveEvent(tplMessageTest4, 100)
-
-	close(testEvents)
-	wg.Wait()
-
-	assert.Equal(t, 2, mockSleeper.callCount)
-	assert.Len(t, receivedEvents, 2)
-	assert.Contains(t, receivedEvents[0].Messages, tplMessageTest2)
-	assert.Contains(t, receivedEvents[0].Messages, tplMessageTest3)
-	assert.Len(t, receivedEvents[0].Messages, 2)
-
-	assert.Contains(t, receivedEvents[1].Messages, tplMessageTest4)
-	assert.Len(t, receivedEvents[1].Messages, 1)
 }

main.go

@@ -80,14 +80,32 @@ func main() {
 
 	dataStore := MakeDataStore(log.New(os.Stderr, "DataStore", log.Lmicroseconds), getPersistenceFilePath())
 
-	Consume(
+	collect := CollectFromChannel[satel.Event]{}
-		SendToMatterbridge(
+
-			SendToTg(Throttle(NotifyViaHTTP(tgEvents, config, &wg, log.New(os.Stderr, "HTTPNotify", log.Lmicroseconds)),
+	filterByLastSeen := MakeFilterByLastSeen(&dataStore)
-				&wg, sleeper, log.New(os.Stderr, "MessageThrottle", log.Lmicroseconds)),
+	filterByTypeOrIndex := MakeFilterByTypeOrIndex(config.AllowedTypes, config.AllowedIndexes)
-				tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tgTpl),
+	convert := MakeConvert(
-			s, config, &wg, log.New(os.Stderr, "SendToMatterbridge", log.Lmicroseconds), ircTpl),
+		func(ev satel.Event) GenericMessage { return GenericMessage{ev.BasicEvents} },
 	)
 
+	notifyViaHttp := MakeNofityViaHTTPSync(config, log.New(os.Stderr, "HTTPNotify", log.Lmicroseconds))
+	throttle := MakeThrottleSync(sleeper, log.New(os.Stderr, "MessageThrottle", log.Lmicroseconds), &wg)
+	sendToTg := MakeSendToTelegramSync(tgSender, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tgTpl)
+	sendToMatterbridge := MakeSendToMatterbridgeSync(s, config,
+		log.New(os.Stderr, "SendToMatterbridge", log.Lmicroseconds),
+		ircTpl)
+
+	collect.Then(filterByLastSeen).
+		Then(filterByTypeOrIndex).
+		Then(convert)
+
+	convert.ConvertTo(notifyViaHttp).
+		Then(throttle).
+		Then(sendToTg).
+		Then(sendToMatterbridge)
+
+	collect.Collect(s.Events, &wg, func() {})
+
 	go CloseSatelOnCtrlC(s, &cleanShutdown)
 
 	closeDebugTools := make(chan interface{})
@@ -95,13 +113,6 @@ func main() {
 		WriteMemoryProfilePeriodically(&wg, log.New(os.Stderr, "DebugTools", log.Lmicroseconds), closeDebugTools)
 	}
 
-	for e := range FilterByTypeOrIndex(
-		FilterByLastSeen(s.Events, &wg, &dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
-		&wg, config.AllowedTypes, config.AllowedIndexes) {
-		logger.Print("Received change from SATEL: ", e)
-		tgEvents <- GenericMessage{e.BasicEvents}
-	}
-
 	logger.Print("Closing...")
 	close(closeDebugTools)
 	close(tgEvents)

sender_sync.go

@@ -0,0 +1,137 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"html/template"
+	"log"
+	"net/http"
+
+	"git.sr.ht/~michalr/go-satel"
+)
+
+type SendToTelegramSync struct {
+	SyncFilterImpl[GenericMessage]
+
+	sender Sender
+	logger *log.Logger
+	tpl    *template.Template
+}
+
+func MakeSendToTelegramSync(sender Sender,
+	logger *log.Logger,
+	tpl *template.Template) *SendToTelegramSync {
+	return &SendToTelegramSync{SyncFilterImpl[GenericMessage]{}, sender, logger, tpl}
+}
+
+func (sendToTg *SendToTelegramSync) Call(msg GenericMessage) {
+	err := sendToTg.sender.Send(msg, sendToTg.tpl)
+	if err != nil {
+		// TODO: handle it better
+		panic(err)
+	}
+
+	sendToTg.CallNext(msg)
+}
+
+type SendToMatterbridgeSync struct {
+	SyncFilterImpl[GenericMessage]
+
+	s      SatelNameGetter
+	config AppConfig
+	logger *log.Logger
+	tpl    *template.Template
+}
+
+func MakeSendToMatterbridgeSync(s SatelNameGetter,
+	config AppConfig,
+	logger *log.Logger,
+	tpl *template.Template) *SendToMatterbridgeSync {
+	return &SendToMatterbridgeSync{SyncFilterImpl[GenericMessage]{}, s, config, logger, tpl}
+}
+
+func (mbridge *SendToMatterbridgeSync) Call(msg GenericMessage) {
+	for _, matterbridgeConfig := range mbridge.config.Matterbridge {
+		body, err := json.Marshal(MatterbridgeMessage{
+			Text:     msg.Format(mbridge.tpl, mbridge.s, mbridge.logger),
+			Username: matterbridgeConfig.Username,
+			Gateway:  matterbridgeConfig.Gateway,
+		})
+		if err != nil {
+			mbridge.logger.Fatal("Could not marshal a JSON message: ", err)
+		}
+		req, err := http.NewRequest(http.MethodPost, matterbridgeConfig.URI, bytes.NewBuffer(body))
+		req.Header["Authorization"] = []string{fmt.Sprint("Bearer ", matterbridgeConfig.Token)}
+		client := http.Client{
+			Timeout: httpTimeout,
+		}
+		res, err := client.Do(req)
+		if err != nil {
+			mbridge.logger.Print("Could not POST ", matterbridgeConfig.URI, ": ", err)
+			return
+		}
+		mbridge.logger.Print("Notified via Matterbridge with result ", res.StatusCode)
+	}
+
+	mbridge.CallNext(msg)
+}
+
+type NotifyViaHTTPSync struct {
+	SyncFilterImpl[GenericMessage]
+
+	config AppConfig
+	logger *log.Logger
+}
+
+func MakeNofityViaHTTPSync(config AppConfig, logger *log.Logger) *NotifyViaHTTPSync {
+	return &NotifyViaHTTPSync{SyncFilterImpl[GenericMessage]{}, config, logger}
+}
+
+func doHttpNotification(url string, logger *log.Logger) {
+
+	if len(url) == 0 {
+		return
+	}
+	req, err := http.NewRequest(http.MethodPost, url, nil)
+
+	client := http.Client{
+		Timeout: httpTimeout,
+	}
+	res, err := client.Do(req)
+	if err != nil {
+		logger.Print("Could not POST ", url, ": ", err)
+		return
+	}
+	logger.Print("Notified via HTTP with result ", res.StatusCode)
+}
+
+func notifyAllHttp(urls []string, logger *log.Logger) {
+	for _, uri := range urls {
+		doHttpNotification(uri, logger)
+	}
+}
+
+func (notifyViaHttp *NotifyViaHTTPSync) Call(msg GenericMessage) {
+inner_arm:
+	for _, basicElement := range msg.Messages {
+		if (basicElement.Index == NotificationPartitionIndex) && (basicElement.Type == satel.ArmedPartition) {
+			if basicElement.Value == ArmedPartition_Armed {
+				notifyAllHttp(notifyViaHttp.config.ArmCallbackUrls, notifyViaHttp.logger)
+			} else {
+				notifyAllHttp(notifyViaHttp.config.DisarmCallbackUrls, notifyViaHttp.logger)
+			}
+			break inner_arm
+		}
+	}
+inner_alarm:
+	for _, basicElement := range msg.Messages {
+		if basicElement.Type == satel.PartitionAlarm {
+			if basicElement.Value == PartitionAlarm_Alarm {
+				notifyAllHttp(notifyViaHttp.config.AlarmCallbackUrls, notifyViaHttp.logger)
+				break inner_alarm
+			}
+		}
+	}
+
+}

sender_sync_test.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"html/template"
+	"io"
+	"log"
+	"os"
+	"testing"
+
+	"git.sr.ht/~michalr/go-satel"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	tgSyncSenderMessageTest1 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: true}
+)
+
+func TestSyncTelegramSender_NoChatIdsWontSendAnything(t *testing.T) {
+	a := assert.New(t)
+	tpl := template.Must(template.New("TelegramMessage").Parse(""))
+	mockBot := MockTgBotAPI{}
+	s := MockSatelNameGetter{"test_name"}
+	sender := TgSender{&mockBot, s, log.New(io.Discard, "", 0), []int64{}}
+
+	tested := MakeSendToTelegramSync(&sender, log.New(io.Discard, "", 0), tpl)
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tgSenderMessageTest1}})
+	a.Equal(0, len(mockBot.messages))
+}
+
+func TestSyncTelegramSender_SendsMessageToAllChatIds(t *testing.T) {
+	a := assert.New(t)
+	tpl := template.Must(template.New("TelegramMessage").Parse(""))
+	mockBot := MockTgBotAPI{}
+	s := MockSatelNameGetter{"test_name"}
+	sender := TgSender{&mockBot, s, log.New(os.Stderr, "", 0), []int64{1, 2, 3}}
+
+	tested := MakeSendToTelegramSync(&sender, log.New(os.Stderr, "", 0), tpl)
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tgSenderMessageTest1}})
+	a.Equal(3, len(mockBot.messages))
+}

sender_worker.go

@@ -1,16 +1,8 @@
 package main
 
 import (
-	"bytes"
-	"encoding/json"
-	"fmt"
 	"html/template"
-	"log"
-	"net/http"
-	"sync"
 	"time"
-
-	"git.sr.ht/~michalr/go-satel"
 )
 
 const (
@@ -26,136 +18,8 @@ type Sleeper interface {
 	Sleep(ch chan<- interface{})
 }
 
-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)
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer close(returnEvents)
-
-		for e := range events {
-			returnEvents <- e
-			err := s.Send(e, tpl)
-			if err != nil {
-				// TODO: handle it better
-				panic(err)
-			}
-		}
-	}()
-
-	return returnEvents
-}
-
-func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
-	wg.Add(1)
-	defer wg.Done()
-
-	if len(url) == 0 {
-		return
-	}
-	req, err := http.NewRequest(http.MethodPost, url, nil)
-
-	client := http.Client{
-		Timeout: httpTimeout,
-	}
-	res, err := client.Do(req)
-	if err != nil {
-		logger.Print("Could not POST ", url, ": ", err)
-		return
-	}
-	logger.Print("Notified via HTTP with result ", res.StatusCode)
-}
-
-func notifyAllHttp(urls []string, logger *log.Logger, wg *sync.WaitGroup) {
-	for _, uri := range urls {
-		go doHttpNotification(uri, logger, wg)
-	}
-}
-
-func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage {
-	returnEvents := make(chan GenericMessage)
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer close(returnEvents)
-
-		for e := range events {
-			returnEvents <- e
-		inner_arm:
-			for _, basicElement := range e.Messages {
-				if (basicElement.Index == NotificationPartitionIndex) && (basicElement.Type == satel.ArmedPartition) {
-					if basicElement.Value == ArmedPartition_Armed {
-						notifyAllHttp(config.ArmCallbackUrls, logger, wg)
-					} else {
-						notifyAllHttp(config.DisarmCallbackUrls, logger, wg)
-					}
-					break inner_arm
-				}
-			}
-		inner_alarm:
-			for _, basicElement := range e.Messages {
-				if basicElement.Type == satel.PartitionAlarm {
-					if basicElement.Value == PartitionAlarm_Alarm {
-						notifyAllHttp(config.AlarmCallbackUrls, logger, wg)
-						break inner_alarm
-					}
-				}
-			}
-
-		}
-	}()
-
-	return returnEvents
-}
-
 type MatterbridgeMessage struct {
 	Text     string `json:"text"`
 	Username string `json:"username"`
 	Gateway  string `json:"gateway"`
 }
-
-func SendToMatterbridge(events <-chan GenericMessage, s SatelNameGetter, config AppConfig, wg *sync.WaitGroup, logger *log.Logger, tpl *template.Template) <-chan GenericMessage {
-	returnEvents := make(chan GenericMessage)
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer close(returnEvents)
-
-		for e := range events {
-			returnEvents <- e
-			for _, matterbridgeConfig := range config.Matterbridge {
-				body, err := json.Marshal(MatterbridgeMessage{
-					Text:     e.Format(tpl, s, logger),
-					Username: matterbridgeConfig.Username,
-					Gateway:  matterbridgeConfig.Gateway,
-				})
-				if err != nil {
-					logger.Fatal("Could not marshal a JSON message: ", err)
-				}
-				req, err := http.NewRequest(http.MethodPost, matterbridgeConfig.URI, bytes.NewBuffer(body))
-				req.Header["Authorization"] = []string{fmt.Sprint("Bearer ", matterbridgeConfig.Token)}
-				client := http.Client{
-					Timeout: httpTimeout,
-				}
-				res, err := client.Do(req)
-				if err != nil {
-					logger.Print("Could not POST ", matterbridgeConfig.URI, ": ", err)
-					return
-				}
-				logger.Print("Notified via Matterbridge with result ", res.StatusCode)
-			}
-		}
-	}()
-
-	return returnEvents
-}

telegram_utils_test.go

@@ -7,19 +7,9 @@ import (
 	"testing"
 
 	"git.sr.ht/~michalr/go-satel"
-	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
 	"github.com/stretchr/testify/assert"
 )
 
-type MockTgBotAPI struct {
-	callCount int
-}
-
-func (self *MockTgBotAPI) Send(c tgbotapi.Chattable) (tgbotapi.Message, error) {
-	self.callCount += 1
-	return tgbotapi.Message{}, nil
-}
-
 var (
 	tgSenderMessageTest1 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: true}
 )
@@ -27,9 +17,9 @@ var (
 func TestTelegramSender_NoChatIdsWontSendAnything(t *testing.T) {
 	a := assert.New(t)
 	tpl := template.Must(template.New("TelegramMessage").Parse(""))
-	mockBot := MockTgBotAPI{0}
+	mockBot := MockTgBotAPI{}
 
 	tested := TgSender{&mockBot, MockSatelNameGetter{"mockPart"}, log.New(io.Discard, "", 0), []int64{}}
 	tested.Send(GenericMessage{[]satel.BasicEventElement{tgSenderMessageTest1}}, tpl)
-	a.Equal(0, mockBot.callCount)
+	a.Equal(0, len(mockBot.messages))
 }

templates_test.go

@@ -4,7 +4,6 @@ import (
 	"html/template"
 	"io"
 	"log"
-	"sync"
 	"testing"
 
 	"git.sr.ht/~michalr/go-satel"
@@ -21,49 +20,27 @@ func (self *MockTemplateSender) Send(msg GenericMessage, tpl *template.Template)
 	return nil
 }
 
-type MockSatelNameGetter struct {
-	name string
-}
-
-func (self MockSatelNameGetter) GetName(devType satel.DeviceType, index byte) (*satel.NameEvent, error) {
-	retval := satel.NameEvent{
-		DevType:         devType,
-		DevNumber:       index,
-		DevTypeFunction: 0,
-		DevName:         self.name,
-	}
-	return &retval, nil
-}
-
 var (
 	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) {
-	testEvents := make(chan GenericMessage)
-	wg := sync.WaitGroup{}
 	mockSender := MockTemplateSender{s: MockSatelNameGetter{"mockPart"}}
 	tpl, err := template.New("TestTemplate").Parse(TelegramMessageTemplate)
 	assert.NoError(t, err)
-	Consume(SendToTg(testEvents, &mockSender, &wg, log.New(io.Discard, "", log.Ltime), tpl))
+	tgSender := MakeSendToTelegramSync(&mockSender, log.New(io.Discard, "", log.Ltime), tpl)
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1, tplMessageTest2}}
+	tgSender.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest1, tplMessageTest2}})
-	close(testEvents)
-	wg.Wait()
 
 	// assert.Equal(t, "siemka", mockSender.message)
 }
 
 func TestIRCTemplate(t *testing.T) {
-	testEvents := make(chan GenericMessage)
-	wg := sync.WaitGroup{}
 	mockSender := MockTemplateSender{s: MockSatelNameGetter{"mockPart"}}
-	tpl, err := template.New("TestIRCTemplate").Parse(IRCMessageTemplate)
+	tpl, err := template.New("TestTemplate").Parse(IRCMessageTemplate)
 	assert.NoError(t, err)
-	Consume(SendToTg(testEvents, &mockSender, &wg, log.New(io.Discard, "", log.Ltime), tpl))
+	tgSender := MakeSendToTelegramSync(&mockSender, log.New(io.Discard, "", log.Ltime), tpl)
-	testEvents <- GenericMessage{[]satel.BasicEventElement{tplMessageTest1, tplMessageTest2}}
+	tgSender.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest1, tplMessageTest2}})
-	close(testEvents)
-	wg.Wait()
 
 	// assert.Equal(t, "siemka", mockSender.message)
 }

test_utils.go

@@ -1,9 +1,39 @@
 package main
 
-import "git.sr.ht/~michalr/go-satel"
+import (
+	"git.sr.ht/~michalr/go-satel"
+	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
+)
 
 func makeTestSatelEvent(changeType satel.ChangeType, index int, val bool) satel.Event {
 	return satel.Event{
 		BasicEvents: []satel.BasicEventElement{{Type: changeType, Index: index, Value: val}},
 	}
 }
+
+func makeGenericMessage(changeType satel.ChangeType, index int, val bool) GenericMessage {
+	return GenericMessage{[]satel.BasicEventElement{{Type: changeType, Index: index, Value: val}}}
+}
+
+type MockTgBotAPI struct {
+	messages []tgbotapi.Chattable
+}
+
+func (self *MockTgBotAPI) Send(c tgbotapi.Chattable) (tgbotapi.Message, error) {
+	self.messages = append(self.messages, c)
+	return tgbotapi.Message{}, nil
+}
+
+type MockSatelNameGetter struct {
+	name string
+}
+
+func (self MockSatelNameGetter) GetName(devType satel.DeviceType, index byte) (*satel.NameEvent, error) {
+	retval := satel.NameEvent{
+		DevType:         devType,
+		DevNumber:       index,
+		DevTypeFunction: 0,
+		DevName:         self.name,
+	}
+	return &retval, nil
+}