Merge 7238389 into 998e374

Author: AnalogJ

pkg/notify/amqp.go

@@ -2,34 +2,139 @@ package notify
 
 import (
 	"encoding/json"
-	"fmt"
+	"errors"
 	"github.com/analogj/lodestone-publisher/pkg/model"
 	"github.com/streadway/amqp"
+	"log"
+	"os"
+	"time"
 )
 
+//Based on https://github.com/streadway/amqp/blob/master/example_client_test.go
+
 type AmqpNotify struct {
+	logger   *log.Logger
 	client   *amqp.Connection
 	channel  *amqp.Channel
 	exchange string
 	queue    string
+
+	done            chan bool
+	notifyConnClose chan *amqp.Error
+	notifyChanClose chan *amqp.Error
+	notifyConfirm   chan amqp.Confirmation
+	isReady         bool
 }
 
+const (
+	// When reconnecting to the server after connection failure
+	reconnectDelay = 5 * time.Second
+
+	// When setting up the channel after a channel exception
+	reInitDelay = 2 * time.Second
+
+	// When resending messages the server didn't confirm
+	resendDelay = 5 * time.Second
+)
+
+var (
+	errNotConnected  = errors.New("not connected to a server")
+	errAlreadyClosed = errors.New("already closed: not connected to the server")
+	errShutdown      = errors.New("session is shutting down")
+)
+
 func (n *AmqpNotify) Init(config map[string]string) error {
 	n.exchange = config["exchange"]
 	n.queue = config["queue"]
+	n.logger = log.New(os.Stdout, "", log.LstdFlags)
+
+	go n.handleReconnect(config["amqp-url"])
+	return nil
+}
+
+// handleReconnect will wait for a connection error on
+// notifyConnClose, and then continuously attempt to reconnect.
+func (n *AmqpNotify) handleReconnect(addr string) {
+	for {
+		n.isReady = false
+		log.Println("Attempting to connect")
+
+		conn, err := n.connect(addr)
+
+		if err != nil {
+			log.Println("Failed to connect. Retrying...")
+
+			select {
+			case <-n.done:
+				return
+			case <-time.After(reconnectDelay):
+			}
+			continue
+		}
+
+		if done := n.handleReInit(conn); done {
+			break
+		}
+	}
+}
+
+// connect will create a new AMQP connection
+func (n *AmqpNotify) connect(addr string) (*amqp.Connection, error) {
+	conn, err := amqp.Dial(addr)
+
+	if err != nil {
+		return nil, err
+	}
+
+	n.changeClient(conn)
+	log.Println("Connected!")
+	return conn, nil
+}
+
+// handleReconnect will wait for a channel error
+// and then continuously attempt to re-initialize both channels
+func (n *AmqpNotify) handleReInit(conn *amqp.Connection) bool {
+	for {
+		n.isReady = false
+
+		err := n.init(conn)
+
+		if err != nil {
+			log.Println("Failed to initialize channel. Retrying...")
+
+			select {
+			case <-n.done:
+				return true
+			case <-time.After(reInitDelay):
+			}
+			continue
+		}
+
+		select {
+		case <-n.done:
+			return true
+		case <-n.notifyConnClose:
+			log.Println("Connection closed. Reconnecting...")
+			return false
+		case <-n.notifyChanClose:
+			log.Println("Channel closed. Re-running init...")
+		}
+	}
+}
+
+// init will initialize channel & declare queue
+func (n *AmqpNotify) init(conn *amqp.Connection) error {
+	ch, err := conn.Channel()
 
-	client, err := amqp.Dial(config["amqp-url"])
 	if err != nil {
 		return err
 	}
-	n.client = client
 
-	//test connection
-	ch, err := client.Channel()
+	err = ch.Confirm(false)
+
 	if err != nil {
 		return err
 	}
-	n.channel = ch
 
 	err = ch.ExchangeDeclare(
 		n.exchange,
@@ -56,35 +161,104 @@ func (n *AmqpNotify) Init(config map[string]string) error {
 		return err
 	}
 
-	return err
+	n.changeChannel(ch)
+	n.isReady = true
+	log.Println("Setup!")
+
+	return nil
 }
 
+// changeClient takes a new connection to the queue,
+// and updates the close listener to reflect this.
+func (n *AmqpNotify) changeClient(client *amqp.Connection) {
+	n.client = client
+	n.notifyConnClose = make(chan *amqp.Error)
+	n.client.NotifyClose(n.notifyConnClose)
+}
+
+// changeChannel takes a new channel to the queue,
+// and updates the channel listeners to reflect this.
+func (n *AmqpNotify) changeChannel(channel *amqp.Channel) {
+	n.channel = channel
+	n.notifyChanClose = make(chan *amqp.Error)
+	n.notifyConfirm = make(chan amqp.Confirmation, 1)
+	n.channel.NotifyClose(n.notifyChanClose)
+	n.channel.NotifyPublish(n.notifyConfirm)
+}
+
+// Publish will push data onto the queue, and wait for a confirm.
+// If no confirms are received until within the resendTimeout,
+// it continuously re-sends messages until a confirm is received.
+// This will block until the server sends a confirm. Errors are
+// only returned if the push action itself fails, see UnsafePush.
 func (n *AmqpNotify) Publish(event model.S3Event) error {
-	fmt.Println("Publishing event..")
+	if !n.isReady {
+		return errors.New("failed to publish event: not connected")
+	}
+
+	n.logger.Println("Publishing event..")
 
 	b, err := json.Marshal(event)
 	if err != nil {
 		return err
 	}
 
+	for {
+		err := n.unsafePublish(b)
+		if err != nil {
+			n.logger.Println("Publish failed. Retrying...")
+			select {
+			case <-n.done:
+				return errShutdown
+			case <-time.After(resendDelay):
+			}
+			continue
+		}
+		select {
+		case confirm := <-n.notifyConfirm:
+			if confirm.Ack {
+				n.logger.Println("Publish confirmed!")
+				return nil
+			}
+		case <-time.After(resendDelay):
+		}
+		n.logger.Println("Publish didn't confirm. Retrying...")
+	}
+}
+
+// UnsafePush will push to the queue without checking for
+// confirmation. It returns an error if it fails to connect.
+// No guarantees are provided for whether the server will
+// recieve the message.
+func (n *AmqpNotify) unsafePublish(data []byte) error {
+	if !n.isReady {
+		return errNotConnected
+	}
 	return n.channel.Publish(
 		n.exchange, // exchange
 		n.queue,    // routing key
-		false,      // mandatory
-		false,      // immediate
+		false,      // Mandatory
+		false,      // Immediate
 		amqp.Publishing{
 			ContentType: "application/json",
-			Body:        b,
+			Body:        data,
 		},
 	)
 }
 
 func (n *AmqpNotify) Close() error {
-	if err := n.client.Close(); err != nil {
+	if !n.isReady {
+		return errAlreadyClosed
+	}
+	err := n.channel.Close()
+	if err != nil {
 		return err
 	}
-	if err := n.channel.Close(); err != nil {
+	err = n.client.Close()
+	if err != nil {
 		return err
 	}
+	close(n.done)
+	n.isReady = false
 	return nil
 }

pkg/watch/fs.go

@@ -28,6 +28,7 @@ func (fs *FsWatcher) Start(notifyClient notify.Interface, config map[string]stri
 		fmt.Println("ERROR", err)
 	}
 
+	fmt.Println("Start watching for filesystem events")
 	done := make(chan bool)
 
 	go func() {