home_data/gatherer/piper.go

40 lines
920 B
Go

package main
import (
"github.com/influxdata/influxdb-client-go/v2/api/write"
amqp "github.com/rabbitmq/amqp091-go"
)
type Piper struct {
influx *InfluxWriter
delivery <-chan amqp.Delivery
}
func (p *Piper) Init(writer *InfluxWriter, amqp AMQPConnection, queue_name string, routing_key string) {
// Create a new client using an InfluxDB server base URL and an authentication token
p.influx = writer
p.delivery, _ = amqp.ConnectToQueue(queue_name, routing_key)
notify := amqp.NotifyError()
go func() {
for _ = range notify {
p.delivery, _ = amqp.ConnectToQueue(queue_name, routing_key)
println("Reconnected to queue")
}
}()
}
func (p *Piper) RunPipe(
conversion func(amqp.Delivery) ([]*write.Point, error),
) {
// Create point using fluent style
for msg := range p.delivery {
points, _ := conversion(msg)
// Write data
for _, point := range points {
p.influx.Write(point)
}
}
}