Synchronous HTTP notification

filters_sync.go

@@ -1,34 +1,95 @@
 package main
 
-import "sync"
+import (
+	"log"
+	"sync"
+)
 
-type SyncFilter interface {
+type SyncFilter[MsgType any] interface {
-	Then(what SyncFilter) SyncFilter
+	Then(what SyncFilter[MsgType]) SyncFilter[MsgType]
-	Call(msg GenericMessage)
+	Call(msg MsgType)
 }
 
-type CollectFromChannel struct {
+type SyncFilterImpl[MsgType any] struct {
-	next SyncFilter
+	next SyncFilter[MsgType]
 }
 
-func (self *CollectFromChannel) Then(what SyncFilter) SyncFilter {
+func (impl *SyncFilterImpl[MsgType]) Then(what SyncFilter[MsgType]) SyncFilter[MsgType] {
-	self.next = what
+	impl.next = what
 	return what
 }
 
-func (self *CollectFromChannel) Call(msg GenericMessage) {
+func (impl *SyncFilterImpl[MsgType]) CallNext(msg MsgType) {
-	if self.next != nil {
+	if impl.next != nil {
-		self.next.Call(msg)
+		impl.next.Call(msg)
 	}
 }
 
-func (self CollectFromChannel) Collect(events <-chan GenericMessage, wg *sync.WaitGroup) {
+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 onClose()
 
 		for e := range events {
-			self.Call(e)
+			collect.Call(e)
 		}
 	}()
 }
+
+type ThrottleSync struct {
+	SyncFilterImpl[GenericMessage]
+
+	events chan GenericMessage
+}
+
+func MakeThrottleSync(sleeper Sleeper, logger *log.Logger, wg *sync.WaitGroup) *ThrottleSync {
+	events := make(chan GenericMessage)
+
+	throttle := ThrottleSync{SyncFilterImpl[GenericMessage]{}, events}
+
+	wg.Add(1)
+	go func() {
+		timeoutEvents := make(chan interface{})
+		var currentEvent *GenericMessage = nil
+	loop:
+		for {
+			select {
+			case ev, ok := <-events:
+				if !ok {
+					break loop
+				}
+				if currentEvent == nil {
+					logger.Print("Waiting for more messages to arrive before sending...")
+					sleeper.Sleep(timeoutEvents)
+				}
+				currentEvent = appendToGenericMessage(currentEvent, &ev)
+			case <-timeoutEvents:
+				logger.Print("Time's up, sending all messages we've got for now.")
+				throttle.CallNext(*currentEvent)
+				currentEvent = nil
+			}
+		}
+
+		// If anything is left to be sent, send it now
+		if currentEvent != nil {
+			throttle.CallNext(*currentEvent)
+		}
+		wg.Done()
+		logger.Print("Throttling goroutine finishing")
+	}()
+
+	return &throttle
+}
+
+func (throttle *ThrottleSync) Close() { close(throttle.events) }
+
+func (throttle *ThrottleSync) Call(msg GenericMessage) {
+	throttle.events <- msg
+}

filters_test.go

@@ -332,33 +332,27 @@ func TestThrottle_ManyMessagesInOneEvent(t *testing.T) {
 }
 
 type SyncMockFilter struct {
+	SyncFilterImpl[GenericMessage]
 	collected []GenericMessage
-	next      SyncFilter
-}
-
-func (self *SyncMockFilter) Then(what SyncFilter) SyncFilter {
-	self.next = what
-	return what
 }
 
 func (self *SyncMockFilter) Call(msg GenericMessage) {
 	self.collected = append(self.collected, msg)
-	if self.next != nil {
+	self.CallNext(msg)
-		self.next.Call(msg)
-	}
 }
 
 func TestSyncCollect(t *testing.T) {
 	testEvents := make(chan GenericMessage)
 	wg := sync.WaitGroup{}
 
-	tested := CollectFromChannel{}
+	tested := CollectFromChannel[GenericMessage]{}
 	mock := &SyncMockFilter{}
 	mock2 := &SyncMockFilter{}
 
 	tested.Then(mock).Then(mock2)
 
-	tested.Collect(testEvents, &wg)
+	wg.Add(1)
+	tested.Collect(testEvents, &wg, func() { wg.Done() })
 
 	testEvents <- makeGenericMessage(satel.ArmedPartition, 1, true)
 	testEvents <- makeGenericMessage(satel.DoorOpened, 2, true)
@@ -386,3 +380,40 @@ func TestSyncCollect(t *testing.T) {
 	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}
+
+	var (
+		tplMessageTest1 = satel.BasicEventElement{Type: satel.ArmedPartition, Index: 1, Value: true}
+		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}
+	)
+
+	tested := MakeThrottleSync(&mockSleeper, fakeLog, &wg)
+	mock := &SyncMockFilter{}
+
+	tested.Then(mock)
+
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest1}})
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest2}})
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest3}})
+	*mockSleeper.ch <- nil
+
+	tested.Call(GenericMessage{[]satel.BasicEventElement{tplMessageTest4}})
+
+	tested.Close()
+	wg.Wait()
+
+	assert.Equal(t, 2, mockSleeper.callCount)
+	assert.Len(t, mock.collected, 2)
+	assert.Contains(t, mock.collected[0].Messages, tplMessageTest2)
+	assert.Contains(t, mock.collected[0].Messages, tplMessageTest3)
+	assert.Len(t, mock.collected[0].Messages, 2)
+
+	assert.Contains(t, mock.collected[1].Messages, tplMessageTest4)
+	assert.Len(t, mock.collected[1].Messages, 1)
+}

sender_sync.go

@@ -0,0 +1,113 @@
+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 (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

@@ -54,9 +54,7 @@ func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger
 	return returnEvents
 }
 
-func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
+func doHttpNotification(url string, logger *log.Logger) {
-	wg.Add(1)
-	defer wg.Done()
 
 	if len(url) == 0 {
 		return
@@ -74,9 +72,9 @@ func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
 	logger.Print("Notified via HTTP with result ", res.StatusCode)
 }
 
-func notifyAllHttp(urls []string, logger *log.Logger, wg *sync.WaitGroup) {
+func notifyAllHttp(urls []string, logger *log.Logger) {
 	for _, uri := range urls {
-		go doHttpNotification(uri, logger, wg)
+		go doHttpNotification(uri, logger)
 	}
 }
 
@@ -94,9 +92,9 @@ func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.Wait
 			for _, basicElement := range e.Messages {
 				if (basicElement.Index == NotificationPartitionIndex) && (basicElement.Type == satel.ArmedPartition) {
 					if basicElement.Value == ArmedPartition_Armed {
-						notifyAllHttp(config.ArmCallbackUrls, logger, wg)
+						notifyAllHttp(config.ArmCallbackUrls, logger)
 					} else {
-						notifyAllHttp(config.DisarmCallbackUrls, logger, wg)
+						notifyAllHttp(config.DisarmCallbackUrls, logger)
 					}
 					break inner_arm
 				}
@@ -105,7 +103,7 @@ func NotifyViaHTTP(events <-chan GenericMessage, config AppConfig, wg *sync.Wait
 			for _, basicElement := range e.Messages {
 				if basicElement.Type == satel.PartitionAlarm {
 					if basicElement.Value == PartitionAlarm_Alarm {
-						notifyAllHttp(config.AlarmCallbackUrls, logger, wg)
+						notifyAllHttp(config.AlarmCallbackUrls, logger)
 						break inner_alarm
 					}
 				}

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

@@ -21,20 +21,6 @@ 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}

test_utils.go

@@ -1,6 +1,9 @@
 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{
@@ -11,3 +14,26 @@ func makeTestSatelEvent(changeType satel.ChangeType, index int, val bool) satel.
 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
+}