by ref

gatherer/connections.go

@@ -99,6 +99,7 @@ func (amqpConn *AMQPConnection) ConnectToQueue(name string, routing_key string)
 func (amqpConn *AMQPConnection) NotifyError() <-chan bool {
 	notify := make(chan bool)
 	amqpConn.observers = append(amqpConn.observers, notify)
+	log.Println("Added observer")
 	return notify
 }
 

gatherer/main.go

@@ -53,11 +53,11 @@ func main() {
 	var forever chan struct{}
 
 	shellPipe := Piper{}
-	shellPipe.Init(&influx, amqpConn, "shelly", "shellies.#")
+	shellPipe.Init(&influx, &amqpConn, "shelly", "shellies.#")
 	go shellPipe.RunPipe(ShellyConverter)
 
 	tasmotaPipe := Piper{}
-	tasmotaPipe.Init(&influx, amqpConn, "tasmota", "tele.#.SENSOR")
+	tasmotaPipe.Init(&influx, &amqpConn, "tasmota", "tele.#.SENSOR")
 	go tasmotaPipe.RunPipe(TasmotaConverter)
 
 	log.Printf(" [*] Waiting for logs. To exit press CTRL+C")

gatherer/piper.go

@@ -10,7 +10,7 @@ type Piper struct {
 	delivery <-chan amqp.Delivery
 }
 
-func (p *Piper) Init(writer *InfluxWriter, amqp AMQPConnection, queue_name string, routing_key string) {
+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)