Potentially multiple callback URLs for events
This commit is contained in:
parent
2c3286206e
commit
12fc45ec38
84
config.go
84
config.go
|
@ -1,6 +1,12 @@
|
||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"flag"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"os"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.sr.ht/~michalr/go-satel"
|
"git.sr.ht/~michalr/go-satel"
|
||||||
|
@ -12,4 +18,82 @@ type AppConfig struct {
|
||||||
allowedTypes []satel.ChangeType
|
allowedTypes []satel.ChangeType
|
||||||
allowedIndexes []int
|
allowedIndexes []int
|
||||||
poolInterval time.Duration
|
poolInterval time.Duration
|
||||||
|
armCallbackUrls []string
|
||||||
|
disarmCallbackUrls []string
|
||||||
|
alarmCallbackUrls []string
|
||||||
|
}
|
||||||
|
|
||||||
|
func getCmdLineParams(config *AppConfig, logger *log.Logger) {
|
||||||
|
satelApiAddr := flag.String("satel-addr", "", "Address that should be used to connect to the SATEL device")
|
||||||
|
satelApiPort := flag.String("satel-port", "7094", "Port that should be used to connect to the SATEL device")
|
||||||
|
chatIdRaw := flag.String("tg-chat-id", "", "Telegram Chat ID where to send updates. Use \",\" to specify multiple IDs.")
|
||||||
|
allowedTypesRaw := flag.String("allowed-types", "", "Satel change types that are allowed. All other types will be discarded. By default all are allowed. Use \",\" to specify multiple types.")
|
||||||
|
allowedIndexesRaw := flag.String("allowed-indexes", "", "Satel indexes (zones?) that are allowed. All other indexes will be discarded. By default all are allowed. Use \",\" to specify multiple indexes.")
|
||||||
|
satelPoolInterval := flag.Duration("pool-interval", 5*time.Second, "How often should the SATEL device be pooled for changes? Default: 5 seconds.")
|
||||||
|
flag.Parse()
|
||||||
|
|
||||||
|
if len(*satelApiAddr) == 0 || len(*satelApiPort) == 0 || len(*chatIdRaw) == 0 {
|
||||||
|
logger.Fatal("Use --satel-addr=ADDR, --satel-port=PORT and --tg-chat-id=CHAT_ID command line flags to continue.")
|
||||||
|
}
|
||||||
|
chatIdsStrings := strings.Split(*chatIdRaw, ",")
|
||||||
|
var chatIds []int64
|
||||||
|
for _, chatIdStr := range chatIdsStrings {
|
||||||
|
chatId, err := strconv.ParseInt(chatIdStr, 10, 64)
|
||||||
|
if err != nil {
|
||||||
|
logger.Fatalf("Tried to use a non-int value for one of tg_chatIds: %s. That's bad.", chatIdStr)
|
||||||
|
}
|
||||||
|
chatIds = append(chatIds, chatId)
|
||||||
|
}
|
||||||
|
allowedTypesStrings := strings.Split(*allowedTypesRaw, ",")
|
||||||
|
var allowedTypes []satel.ChangeType
|
||||||
|
for _, allowedTypeStr := range allowedTypesStrings {
|
||||||
|
if len(allowedTypeStr) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
allowedType, err := StringToSatelChangeType(allowedTypeStr)
|
||||||
|
if err != nil {
|
||||||
|
logger.Fatalf("Error trying to understand an allowed type: %s.", err)
|
||||||
|
}
|
||||||
|
allowedTypes = append(allowedTypes, allowedType)
|
||||||
|
}
|
||||||
|
allowedIndexesStrings := strings.Split(*allowedIndexesRaw, ",")
|
||||||
|
var allowedIndexes []int
|
||||||
|
for _, allowedIndexStr := range allowedIndexesStrings {
|
||||||
|
if len(allowedIndexStr) == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
allowedIndex, err := strconv.ParseInt(allowedIndexStr, 10, 0)
|
||||||
|
if err != nil {
|
||||||
|
logger.Fatalf("Tried to use a non-int value for one of allowed indexes: %s. That's bad.", allowedIndexStr)
|
||||||
|
}
|
||||||
|
allowedIndexes = append(allowedIndexes, int(allowedIndex))
|
||||||
|
}
|
||||||
|
|
||||||
|
satelAddr := fmt.Sprintf("%s:%s", *satelApiAddr, *satelApiPort)
|
||||||
|
|
||||||
|
config.satelAddr = satelAddr
|
||||||
|
config.chatIds = chatIds
|
||||||
|
config.allowedTypes = allowedTypes
|
||||||
|
config.allowedIndexes = allowedIndexes
|
||||||
|
config.poolInterval = *satelPoolInterval
|
||||||
|
}
|
||||||
|
|
||||||
|
func MakeConfig(logger *log.Logger) AppConfig {
|
||||||
|
config := AppConfig{}
|
||||||
|
config.armCallbackUrls = []string{}
|
||||||
|
config.disarmCallbackUrls = []string{}
|
||||||
|
config.alarmCallbackUrls = []string{}
|
||||||
|
|
||||||
|
if len(os.Getenv("NOTIFY_URL_ARM")) != 0 {
|
||||||
|
config.armCallbackUrls = append(config.armCallbackUrls, os.Getenv("NOTIFY_URL_ARM"))
|
||||||
|
}
|
||||||
|
if len(os.Getenv("NOTIFY_URL_DISARM")) != 0 {
|
||||||
|
config.disarmCallbackUrls = append(config.disarmCallbackUrls, os.Getenv("NOTIFY_URL_DISARM"))
|
||||||
|
}
|
||||||
|
if len(os.Getenv("ALARM_URL_ARM")) != 0 {
|
||||||
|
config.alarmCallbackUrls = append(config.alarmCallbackUrls, os.Getenv("ALARM_URL_ARM"))
|
||||||
|
}
|
||||||
|
|
||||||
|
getCmdLineParams(&config, logger)
|
||||||
|
return config
|
||||||
}
|
}
|
||||||
|
|
66
main.go
66
main.go
|
@ -1,15 +1,11 @@
|
||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"flag"
|
|
||||||
"fmt"
|
|
||||||
"html/template"
|
"html/template"
|
||||||
"log"
|
"log"
|
||||||
"net"
|
"net"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
@ -34,56 +30,6 @@ func (self RealSleeper) Sleep(ch chan<- interface{}) {
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
func getCmdLineParams(logger *log.Logger) AppConfig {
|
|
||||||
satelApiAddr := flag.String("satel-addr", "", "Address that should be used to connect to the SATEL device")
|
|
||||||
satelApiPort := flag.String("satel-port", "7094", "Port that should be used to connect to the SATEL device")
|
|
||||||
chatIdRaw := flag.String("tg-chat-id", "", "Telegram Chat ID where to send updates. Use \",\" to specify multiple IDs.")
|
|
||||||
allowedTypesRaw := flag.String("allowed-types", "", "Satel change types that are allowed. All other types will be discarded. By default all are allowed. Use \",\" to specify multiple types.")
|
|
||||||
allowedIndexesRaw := flag.String("allowed-indexes", "", "Satel indexes (zones?) that are allowed. All other indexes will be discarded. By default all are allowed. Use \",\" to specify multiple indexes.")
|
|
||||||
satelPoolInterval := flag.Duration("pool-interval", 5*time.Second, "How often should the SATEL device be pooled for changes? Default: 5 seconds.")
|
|
||||||
flag.Parse()
|
|
||||||
|
|
||||||
if len(*satelApiAddr) == 0 || len(*satelApiPort) == 0 || len(*chatIdRaw) == 0 {
|
|
||||||
logger.Fatal("Use --satel-addr=ADDR, --satel-port=PORT and --tg-chat-id=CHAT_ID command line flags to continue.")
|
|
||||||
}
|
|
||||||
chatIdsStrings := strings.Split(*chatIdRaw, ",")
|
|
||||||
var chatIds []int64
|
|
||||||
for _, chatIdStr := range chatIdsStrings {
|
|
||||||
chatId, err := strconv.ParseInt(chatIdStr, 10, 64)
|
|
||||||
if err != nil {
|
|
||||||
logger.Fatalf("Tried to use a non-int value for one of tg_chatIds: %s. That's bad.", chatIdStr)
|
|
||||||
}
|
|
||||||
chatIds = append(chatIds, chatId)
|
|
||||||
}
|
|
||||||
allowedTypesStrings := strings.Split(*allowedTypesRaw, ",")
|
|
||||||
var allowedTypes []satel.ChangeType
|
|
||||||
for _, allowedTypeStr := range allowedTypesStrings {
|
|
||||||
if len(allowedTypeStr) == 0 {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
allowedType, err := StringToSatelChangeType(allowedTypeStr)
|
|
||||||
if err != nil {
|
|
||||||
logger.Fatalf("Error trying to understand an allowed type: %s.", err)
|
|
||||||
}
|
|
||||||
allowedTypes = append(allowedTypes, allowedType)
|
|
||||||
}
|
|
||||||
allowedIndexesStrings := strings.Split(*allowedIndexesRaw, ",")
|
|
||||||
var allowedIndexes []int
|
|
||||||
for _, allowedIndexStr := range allowedIndexesStrings {
|
|
||||||
if len(allowedIndexStr) == 0 {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
allowedIndex, err := strconv.ParseInt(allowedIndexStr, 10, 0)
|
|
||||||
if err != nil {
|
|
||||||
logger.Fatalf("Tried to use a non-int value for one of allowed indexes: %s. That's bad.", allowedIndexStr)
|
|
||||||
}
|
|
||||||
allowedIndexes = append(allowedIndexes, int(allowedIndex))
|
|
||||||
}
|
|
||||||
|
|
||||||
satelAddr := fmt.Sprintf("%s:%s", *satelApiAddr, *satelApiPort)
|
|
||||||
return AppConfig{satelAddr, chatIds, allowedTypes, allowedIndexes, *satelPoolInterval}
|
|
||||||
}
|
|
||||||
|
|
||||||
func makeSatel(satelAddr string, poolInterval time.Duration) *satel.Satel {
|
func makeSatel(satelAddr string, poolInterval time.Duration) *satel.Satel {
|
||||||
satelConn, err := net.Dial("tcp", satelAddr)
|
satelConn, err := net.Dial("tcp", satelAddr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -110,10 +56,10 @@ func main() {
|
||||||
)
|
)
|
||||||
|
|
||||||
stopRequested.Store(false)
|
stopRequested.Store(false)
|
||||||
cfg := getCmdLineParams(logger)
|
config := MakeConfig(logger)
|
||||||
|
|
||||||
s := makeSatel(cfg.satelAddr, cfg.poolInterval)
|
s := makeSatel(config.satelAddr, config.poolInterval)
|
||||||
logger.Printf("Connected to Satel: %s", cfg.satelAddr)
|
logger.Printf("Connected to Satel: %s", config.satelAddr)
|
||||||
|
|
||||||
bot, err := tgbotapi.NewBotAPI(os.Getenv("TELEGRAM_APITOKEN"))
|
bot, err := tgbotapi.NewBotAPI(os.Getenv("TELEGRAM_APITOKEN"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -121,14 +67,14 @@ func main() {
|
||||||
}
|
}
|
||||||
logger.Print("Created Telegram Bot API client")
|
logger.Print("Created Telegram Bot API client")
|
||||||
|
|
||||||
tgSender := TgSender{bot, s, log.New(os.Stderr, "TgFormatter", log.Lmicroseconds), cfg.chatIds}
|
tgSender := TgSender{bot, s, log.New(os.Stderr, "TgFormatter", log.Lmicroseconds), config.chatIds}
|
||||||
|
|
||||||
tpl := template.Must(template.New("TelegramMessage").Parse(TelegramMessageTemplate))
|
tpl := template.Must(template.New("TelegramMessage").Parse(TelegramMessageTemplate))
|
||||||
|
|
||||||
dataStore := MakeDataStore(log.New(os.Stderr, "DataStore", log.Lmicroseconds), getPersistenceFilePath())
|
dataStore := MakeDataStore(log.New(os.Stderr, "DataStore", log.Lmicroseconds), getPersistenceFilePath())
|
||||||
|
|
||||||
Consume(
|
Consume(
|
||||||
SendToTg(Throttle(NotifyViaHTTP(tgEvents, &wg, log.New(os.Stderr, "HTTPNotify", log.Lmicroseconds)),
|
SendToTg(Throttle(NotifyViaHTTP(tgEvents, config, &wg, log.New(os.Stderr, "HTTPNotify", log.Lmicroseconds)),
|
||||||
&wg, sleeper, log.New(os.Stderr, "MessageThrottle", log.Lmicroseconds)),
|
&wg, sleeper, log.New(os.Stderr, "MessageThrottle", log.Lmicroseconds)),
|
||||||
tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tpl),
|
tgSender, &wg, log.New(os.Stderr, "SendToTg", log.Lmicroseconds), tpl),
|
||||||
)
|
)
|
||||||
|
@ -138,7 +84,7 @@ func main() {
|
||||||
for stopRequested.Load() == false {
|
for stopRequested.Load() == false {
|
||||||
for e := range FilterByTypeOrIndex(
|
for e := range FilterByTypeOrIndex(
|
||||||
FilterByLastSeen(s.Events, &wg, &dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
|
FilterByLastSeen(s.Events, &wg, &dataStore, log.New(os.Stderr, "FilterByLastSeen", log.Lmicroseconds)),
|
||||||
&wg, cfg.allowedTypes, cfg.allowedIndexes) {
|
&wg, config.allowedTypes, config.allowedIndexes) {
|
||||||
logger.Print("Received change from SATEL: ", e)
|
logger.Print("Received change from SATEL: ", e)
|
||||||
tgEvents <- GenericMessage{e.BasicEvents}
|
tgEvents <- GenericMessage{e.BasicEvents}
|
||||||
}
|
}
|
||||||
|
|
|
@ -4,7 +4,6 @@ import (
|
||||||
"html/template"
|
"html/template"
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"git.sr.ht/~michalr/go-satel"
|
"git.sr.ht/~michalr/go-satel"
|
||||||
|
@ -52,6 +51,7 @@ func SendToTg(events <-chan GenericMessage, s Sender, wg *sync.WaitGroup, logger
|
||||||
func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
|
func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
|
|
||||||
if len(url) == 0 {
|
if len(url) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
@ -65,43 +65,41 @@ func doHttpNotification(url string, logger *log.Logger, wg *sync.WaitGroup) {
|
||||||
logger.Print("Notified via HTTP with result ", res.StatusCode)
|
logger.Print("Notified via HTTP with result ", res.StatusCode)
|
||||||
}
|
}
|
||||||
|
|
||||||
func NotifyViaHTTP(events <-chan GenericMessage, wg *sync.WaitGroup, logger *log.Logger) <-chan GenericMessage {
|
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)
|
returnEvents := make(chan GenericMessage)
|
||||||
armCallbackUrl := os.Getenv("NOTIFY_URL_ARM")
|
|
||||||
disarmCallbackUrl := os.Getenv("NOTIFY_URL_DISARM")
|
|
||||||
alarmCallbackUrl := os.Getenv("ALARM_URL_ARM")
|
|
||||||
armDisarmCallbackEnabled := (len(armCallbackUrl) != 0) && (len(disarmCallbackUrl) != 0)
|
|
||||||
alarmCallbackEnabled := (len(alarmCallbackUrl) != 0)
|
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
for e := range events {
|
for e := range events {
|
||||||
returnEvents <- e
|
returnEvents <- e
|
||||||
if armDisarmCallbackEnabled {
|
|
||||||
inner_arm:
|
inner_arm:
|
||||||
for _, basicElement := range e.Messages {
|
for _, basicElement := range e.Messages {
|
||||||
if (basicElement.Index == NotificationPartitionIndex) && (basicElement.Type == satel.ArmedPartition) {
|
if (basicElement.Index == NotificationPartitionIndex) && (basicElement.Type == satel.ArmedPartition) {
|
||||||
if basicElement.Value == ArmedPartition_Armed {
|
if basicElement.Value == ArmedPartition_Armed {
|
||||||
go doHttpNotification(armCallbackUrl, logger, wg)
|
notifyAllHttp(config.armCallbackUrls, logger, wg)
|
||||||
} else {
|
} else {
|
||||||
go doHttpNotification(disarmCallbackUrl, logger, wg)
|
notifyAllHttp(config.disarmCallbackUrls, logger, wg)
|
||||||
}
|
}
|
||||||
break inner_arm
|
break inner_arm
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
if alarmCallbackEnabled {
|
|
||||||
inner_alarm:
|
inner_alarm:
|
||||||
for _, basicElement := range e.Messages {
|
for _, basicElement := range e.Messages {
|
||||||
if basicElement.Type == satel.PartitionAlarm {
|
if basicElement.Type == satel.PartitionAlarm {
|
||||||
if basicElement.Value == PartitionAlarm_Alarm {
|
if basicElement.Value == PartitionAlarm_Alarm {
|
||||||
go doHttpNotification(alarmCallbackUrl, logger, wg)
|
notifyAllHttp(config.alarmCallbackUrls, logger, wg)
|
||||||
break inner_alarm
|
break inner_alarm
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
}
|
||||||
close(returnEvents)
|
close(returnEvents)
|
||||||
}()
|
}()
|
||||||
|
|
Loading…
Reference in New Issue