Browse Source

add an option to filter out topics matching a regex

main
Nicolas Massé 4 years ago
parent
commit
090a8197a6
  1. 13
      archive.go
  2. 1
      cli/cmd/archive.go
  3. 1
      mqtt-archiver.yaml

13
archive.go

@ -31,6 +31,7 @@ import (
"os" "os"
"path" "path"
"path/filepath" "path/filepath"
"regexp"
"sync" "sync"
"time" "time"
@ -67,6 +68,8 @@ type Archiver struct {
WorkingDir string // location to store JSON files WorkingDir string // location to store JSON files
Logger log.Logger // a logger Logger log.Logger // a logger
SubscribePattern string // the pattern (ie. "#") to subscribe to SubscribePattern string // the pattern (ie. "#") to subscribe to
FilterRegex string // topics matching this regex will filtered out
filter *regexp.Regexp // compiled regex: topics matching this regex will filtered out
currentFilename string // the current JSON filename currentFilename string // the current JSON filename
wg sync.WaitGroup // a wait group to keep track of each running go routine wg sync.WaitGroup // a wait group to keep track of each running go routine
client mqtt.Client // the MQTT client client mqtt.Client // the MQTT client
@ -84,6 +87,7 @@ const (
func (archiver *Archiver) StartArchive() error { func (archiver *Archiver) StartArchive() error {
archiver.done = make(chan bool) archiver.done = make(chan bool)
archiver.eventLog = make(chan EventLogEntry, 10) archiver.eventLog = make(chan EventLogEntry, 10)
archiver.filter = regexp.MustCompile(archiver.FilterRegex)
// connect to MQTT server // connect to MQTT server
SetMqttLogger(&archiver.Logger) SetMqttLogger(&archiver.Logger)
@ -388,12 +392,17 @@ func (archiver *Archiver) openLogFile(fileName string) (*os.File, error) {
// processMessage is the callback routine called by the MQTT library to process // processMessage is the callback routine called by the MQTT library to process
// events. // events.
func (archiver *Archiver) processMessage(c mqtt.Client, m mqtt.Message) { func (archiver *Archiver) processMessage(c mqtt.Client, m mqtt.Message) {
archiver.wg.Add(1)
defer archiver.wg.Done()
if m.Retained() { if m.Retained() {
return return
} }
archiver.wg.Add(1) if archiver.filter.Match([]byte(m.Topic())) {
defer archiver.wg.Done() return
}
entry := EventLogEntry{ entry := EventLogEntry{
Version: 1, Version: 1,
Timestamp: time.Now().UTC(), Timestamp: time.Now().UTC(),

1
cli/cmd/archive.go

@ -80,6 +80,7 @@ var archiveCmd = &cobra.Command{
}, },
SubscribePattern: viper.GetString("subscribePattern"), SubscribePattern: viper.GetString("subscribePattern"),
WorkingDir: viper.GetString("workingDir"), WorkingDir: viper.GetString("workingDir"),
FilterRegex: viper.GetString("exclude"),
Logger: *logger, Logger: *logger,
} }

1
mqtt-archiver.yaml

@ -5,6 +5,7 @@ mqtt:
timeout: 5s timeout: 5s
gracePeriod: 2s gracePeriod: 2s
workingDir: .podman-compose/mqtt-archiver/ workingDir: .podman-compose/mqtt-archiver/
exclude: ^test
s3: s3:
endpoint: localhost:9000 endpoint: localhost:9000
accessKey: dev accessKey: dev

Loading…
Cancel
Save