Skip to content

Commit

Permalink
AMQP auto reconnect feature
Browse files Browse the repository at this point in the history
Closes #207
  • Loading branch information
Eugene Dementiev authored and sparrc committed Sep 24, 2015
1 parent 2452785 commit c6283d1
Showing 1 changed file with 16 additions and 0 deletions.
16 changes: 16 additions & 0 deletions outputs/amqp/amqp.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@ package amqp

import (
"fmt"
"log"
"sync"
"time"

"github.com/influxdb/influxdb/client"
Expand All @@ -18,6 +20,7 @@ type AMQP struct {
RoutingTag string `toml:"routing_tag"`

channel *amqp.Channel
sync.Mutex
}

var sampleConfig = `
Expand All @@ -31,6 +34,8 @@ var sampleConfig = `
`

func (q *AMQP) Connect() error {
q.Lock()
defer q.Unlock()
connection, err := amqp.Dial(q.URL)
if err != nil {
return err
Expand All @@ -53,6 +58,15 @@ func (q *AMQP) Connect() error {
return fmt.Errorf("Failed to declare an exchange: %s", err)
}
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
}

Expand All @@ -69,6 +83,8 @@ func (q *AMQP) Description() string {
}

func (q *AMQP) Write(bp client.BatchPoints) error {
q.Lock()
defer q.Unlock()
if len(bp.Points) == 0 {
return nil
}
Expand Down

0 comments on commit c6283d1

Please sign in to comment.