Switch logging to logrus

client/client.go

@@ -4,12 +4,13 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"log"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
+	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/websocket"
 
 	"github.com/prologic/msgbus"
@@ -67,11 +68,15 @@ func NewClient(url string, options *Options) *Client {
 }
 
 // Handle ...
-func (c *Client) Handle(msg *msgbus.Message) {
-	log.Printf(
-		"[msgbus] received message: id=%d topic=%s payload=%s",
-		msg.ID, msg.Topic.Name, msg.Payload,
-	)
+func (c *Client) Handle(msg *msgbus.Message) error {
+	out, err := json.Marshal(msg)
+	if err != nil {
+		log.Errorf("error marshalling message: %s", err)
+		return err
+	}
+
+	os.Stdout.Write(out)
+	return nil
 }
 
 // Pull ...
@@ -187,7 +192,7 @@ func (s *Subscriber) Run() {
 	for {
 		s.conn, err = websocket.Dial(url, "", origin)
 		if err != nil {
-			log.Printf("error connecting to %s: %s", url, err)
+			log.Warnf("error connecting to %s: %s", url, err)
 			time.Sleep(s.client.reconnect)
 			continue
 		}
@@ -197,11 +202,11 @@ func (s *Subscriber) Run() {
 		select {
 		case err = <-s.errch:
 			if err != nil {
-				log.Printf("lost connection to %s: %s", url, err)
+				log.Warnf("lost connection to %s: %s", url, err)
 				time.Sleep(s.client.reconnect)
 			}
 		case <-s.stopch:
-			log.Printf("shutting down ...")
+			log.Infof("shutting down ...")
 			s.conn.Close()
 			break
 		}

cmd/msgbus/root.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/mitchellh/go-homedir"
+	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
@@ -44,6 +45,11 @@ func init() {
 		"config file (default is $HOME/.msgbus.yaml)",
 	)
 
+	RootCmd.PersistentFlags().BoolP(
+		"debug", "d", false,
+		"Enable debug logging",
+	)
+
 	RootCmd.PersistentFlags().StringP(
 		"uri", "u", "http://localhost:8000",
 		"URI to connect to msgbusd",
@@ -51,6 +57,16 @@ func init() {
 
 	viper.BindPFlag("uri", RootCmd.PersistentFlags().Lookup("uri"))
 	viper.SetDefault("uri", "http://localhost:8000/")
+
+	viper.BindPFlag("debug", RootCmd.PersistentFlags().Lookup("debug"))
+	viper.SetDefault("debug", false)
+
+	// set logging level
+	if viper.GetBool("debug") {
+		log.SetLevel(log.DebugLevel)
+	} else {
+		log.SetLevel(log.InfoLevel)
+	}
 }
 
 // initConfig reads in config file and ENV variables if set.

cmd/msgbusd/main.go

@@ -3,11 +3,12 @@ package main
 import (
 	"flag"
 	"fmt"
-	"log"
 	"net/http"
 	"os"
 	"time"
 
+	log "github.com/sirupsen/logrus"
+
 	"github.com/prologic/msgbus"
 )
 
@@ -15,17 +16,25 @@ func main() {
 
 	var (
 		version bool
+		debug   bool
 		bind    string
 		ttl     time.Duration
 	)
 
 	flag.BoolVar(&version, "v", false, "display version information")
+	flag.BoolVar(&debug, "d", false, "enable debug logging")
 
 	flag.StringVar(&bind, "bind", ":8000", "interface and port to bind to")
 	flag.DurationVar(&ttl, "ttl", 60*time.Second, "default ttl")
 
 	flag.Parse()
 
+	if debug {
+		log.SetLevel(log.DebugLevel)
+	} else {
+		log.SetLevel(log.InfoLevel)
+	}
+
 	if version {
 		fmt.Printf("msgbusd %s", msgbus.FullVersion())
 		os.Exit(0)
@@ -33,6 +42,6 @@ func main() {
 
 	options := msgbus.Options{DefaultTTL: ttl}
 	http.Handle("/", msgbus.NewMessageBus(&options))
-	log.Printf("msgbusd %s listening on %s", msgbus.FullVersion(), bind)
+	log.Infof("msgbusd %s listening on %s", msgbus.FullVersion(), bind)
 	log.Fatal(http.ListenAndServe(bind, nil))
 }

msgbus.go

@@ -3,11 +3,12 @@ package msgbus
 import (
 	"encoding/json"
 	"io/ioutil"
-	"log"
 	"net/http"
 	"strings"
 	"time"
 
+	log "github.com/sirupsen/logrus"
+
 	"golang.org/x/net/websocket"
 )
 
@@ -145,10 +146,10 @@ func (mb *MessageBus) NewMessage(topic *Topic, payload []byte) Message {
 
 // Put ...
 func (mb *MessageBus) Put(message Message) {
-	//log.Printf(
-	//	"[msgbus] PUT id=%d topic=%s payload=%s",
-	//	message.ID, message.Topic.Name, message.Payload,
-	//)
+	log.Debugf(
+		"[msgbus] PUT id=%d topic=%s payload=%s",
+		message.ID, message.Topic.Name, message.Payload,
+	)
 
 	q, ok := mb.queues[message.Topic]
 	if !ok {
@@ -162,7 +163,7 @@ func (mb *MessageBus) Put(message Message) {
 
 // Get ...
 func (mb *MessageBus) Get(topic *Topic) (Message, bool) {
-	//log.Printf("[msgbus] GET topic=%s", topic)
+	log.Debugf("[msgbus] GET topic=%s", topic)
 
 	q, ok := mb.queues[topic]
 	if !ok {
@@ -178,10 +179,10 @@ func (mb *MessageBus) Get(topic *Topic) (Message, bool) {
 
 // NotifyAll ...
 func (mb *MessageBus) NotifyAll(message Message) {
-	//log.Printf(
-	//	"[msgbus] NotifyAll id=%d topic=%s payload=%s",
-	//	message.ID, message.Topic.Name, message.Payload,
-	//)
+	log.Debugf(
+		"[msgbus] NotifyAll id=%d topic=%s payload=%s",
+		message.ID, message.Topic.Name, message.Payload,
+	)
 	ls, ok := mb.listeners[message.Topic]
 	if !ok {
 		return
@@ -191,7 +192,7 @@ func (mb *MessageBus) NotifyAll(message Message) {
 
 // Subscribe ...
 func (mb *MessageBus) Subscribe(id, topic string) chan Message {
-	//log.Printf("[msgbus] Subscribe id=%s topic=%s", id, topic)
+	log.Debugf("[msgbus] Subscribe id=%s topic=%s", id, topic)
 	t, ok := mb.topics[topic]
 	if !ok {
 		t = &Topic{Name: topic, TTL: mb.ttl, Created: time.Now()}
@@ -214,7 +215,7 @@ func (mb *MessageBus) Subscribe(id, topic string) chan Message {
 
 // Unsubscribe ...
 func (mb *MessageBus) Unsubscribe(id, topic string) {
-	//log.Printf("[msgbus] Unsubscribe id=%s topic=%s", id, topic)
+	log.Debugf("[msgbus] Unsubscribe id=%s topic=%s", id, topic)
 	t, ok := mb.topics[topic]
 	if !ok {
 		return
@@ -316,7 +317,7 @@ func (c *Client) Handler() websocket.Handler {
 			err = websocket.JSON.Send(conn, msg)
 			if err != nil {
 				// TODO: Retry? Put the message back in the queue?
-				log.Printf("Error sending msg to %s", c.id)
+				log.Errorf("Error sending msg to %s", c.id)
 				continue
 			}
 		}