dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

AMQP auto reconnect feature

This commit is contained in:
Eugene Dementiev 2015-09-17 22:29:59 +03:00
parent 24527859e6
commit bfddabef56
1 changed files with 16 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
outputs/amqp/amqp.go
View File

@ -2,6 +2,8 @@ package amqp
import ( import (
"fmt" "fmt"
"log"
"sync"
"time" "time"
"github.com/influxdb/influxdb/client" "github.com/influxdb/influxdb/client"
@ -18,6 +20,7 @@ type AMQP struct {
RoutingTag string `toml:"routing_tag"` RoutingTag string `toml:"routing_tag"`
channel *amqp.Channel channel *amqp.Channel
sync.Mutex
} }
var sampleConfig = ` var sampleConfig = `
@ -31,6 +34,8 @@ var sampleConfig = `
` `
func (q *AMQP) Connect() error { func (q *AMQP) Connect() error {
q.Lock()
defer q.Unlock()
connection, err := amqp.Dial(q.URL) connection, err := amqp.Dial(q.URL)
if err != nil { if err != nil {
return err return err
@ -53,6 +58,15 @@ func (q *AMQP) Connect() error {
return fmt.Errorf("Failed to declare an exchange: %s", err) return fmt.Errorf("Failed to declare an exchange: %s", err)
} }
q.channel = channel q.channel = channel
go func() {
log.Printf("Closing: %s", <-connection.NotifyClose(make(chan *amqp.Error)))
log.Printf("Trying to reconnect")
for err := q.Connect(); err != nil; err = q.Connect() {
log.Println(err)
time.Sleep(10 * time.Second)
}
}()
return nil return nil
} }
@ -69,6 +83,8 @@ func (q *AMQP) Description() string {
} }
func (q *AMQP) Write(bp client.BatchPoints) error { func (q *AMQP) Write(bp client.BatchPoints) error {
q.Lock()
defer q.Unlock()
if len(bp.Points) == 0 { if len(bp.Points) == 0 {
return nil return nil
} }