Fix Client API

client/client.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
-	"os"
 	"strings"
 	"sync"
 	"time"
@@ -31,6 +30,8 @@ const (
 	DefaultPingInterval = 60 * time.Second
 )
 
+func noopHandler(msg *msgbus.Message) error { return nil }
+
 // Client ...
 type Client struct {
 	sync.RWMutex
@@ -74,18 +75,6 @@ func NewClient(url string, options *Options) *Client {
 	return client
 }
 
-// Handle ...
-func (c *Client) Handle(msg *msgbus.Message) error {
-	out, err := json.Marshal(msg)
-	if err != nil {
-		return err
-	}
-
-	os.Stdout.Write(out)
-	os.Stdout.Write([]byte{'\r', '\n'})
-	return nil
-}
-
 // Pull ...
 func (c *Client) Pull(topic string) (msg *msgbus.Message, err error) {
 	c.RLock()
@@ -113,9 +102,6 @@ func (c *Client) Pull(topic string) (msg *msgbus.Message, err error) {
 	if err := json.NewDecoder(res.Body).Decode(&msg); err != nil {
 		return nil, err
 	}
-	if err := c.Handle(msg); err != nil {
-		return nil, err
-	}
 
 	return msg, nil
 }
@@ -174,7 +160,7 @@ type Subscriber struct {
 // NewSubscriber ...
 func NewSubscriber(client *Client, topic string, handler msgbus.HandlerFunc) *Subscriber {
 	if handler == nil {
-		handler = client.Handle
+		handler = noopHandler
 	}
 
 	u, err := url.Parse(client.url)

cmd/msgbus/pull.go

@@ -1,6 +1,9 @@
 package main
 
 import (
+	"fmt"
+	"os"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
@@ -38,5 +41,10 @@ func pull(client *client.Client, topic string) {
 		topic = defaultTopic
 	}
 
-	client.Pull(topic)
+	msg, err := client.Pull(topic)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "error reading message: %s\n", err)
+		os.Exit(2)
+	}
+	fmt.Printf("%s\n", msg.Payload)
 }