public API refactoring

README.md

@@ -2,12 +2,15 @@
 
 `go-nsq` is the official Go package for [NSQ][nsq].
 
-The latest stable release is **[0.3.7](https://github.com/bitly/go-nsq/releases/tag/v0.3.7)**
+NOTE: The public API was improved in [pull request #30][pr30] and is not backwards compatible with
+previous releases. **[0.3.7][legacy]** is the last stable release compatible with the legacy API.
 
-[![Build Status](https://secure.travis-ci.org/bitly/go-nsq.png?branch=master)](http://travis-ci.org/bitly/go-nsq)
+The latest stable release is **[0.3.7][latest_tag]**
 
-It provides high-level [Reader][reader] and [Writer][writer] types to implement consumers and
-producers as well as low-level functions to communicate over the [NSQ protocol][protocol].
+[![Build Status](https://secure.travis-ci.org/bitly/go-nsq.png?branch=master)][travis]
+
+It provides high-level [Consumer][consumer] and [Producer][producer] types as well as low-level
+functions to communicate over the [NSQ protocol][protocol].
 
 See the [main repo apps][apps] directory for examples of clients built using this package.
 
@@ -26,5 +29,9 @@ See [godoc][nsq_gopkgdoc] for pretty documentation or:
 [nsq_gopkgdoc]: http://godoc.org/github.com/bitly/go-nsq
 [protocol]: http://bitly.github.io/nsq/clients/tcp_protocol_spec.html
 [apps]: https://github.com/bitly/nsq/tree/master/apps
-[reader]: http://godoc.org/github.com/bitly/go-nsq#Reader
-[writer]: http://godoc.org/github.com/bitly/go-nsq#Writer
+[consumer]: http://godoc.org/github.com/bitly/go-nsq#Consumer
+[producer]: http://godoc.org/github.com/bitly/go-nsq#Producer
+[pr30]: https://github.com/bitly/go-nsq/pull/30
+[legacy]: https://github.com/bitly/go-nsq/releases/tag/v0.3.7
+[travis]: http://travis-ci.org/bitly/go-nsq
+[latest_tag]: https://github.com/bitly/go-nsq/releases/tag/v0.3.7

api_request.go

@@ -26,8 +26,7 @@ func (c *deadlinedConn) Write(b []byte) (n int, err error) {
 	return c.Conn.Write(b)
 }
 
-// A custom http.Transport with support for deadline timeouts
-func NewDeadlineTransport(timeout time.Duration) *http.Transport {
+func newDeadlineTransport(timeout time.Duration) *http.Transport {
 	transport := &http.Transport{
 		Dial: func(netw, addr string) (net.Conn, error) {
 			c, err := net.DialTimeout(netw, addr, timeout)
@@ -40,12 +39,8 @@ func NewDeadlineTransport(timeout time.Duration) *http.Transport {
 	return transport
 }
 
-// ApiRequest is a helper function to perform an HTTP request
-// and parse our NSQ daemon's expected response format, with deadlines.
-//
-//     {"status_code":200, "status_txt":"OK", "data":{...}}
-func ApiRequest(endpoint string) (*simplejson.Json, error) {
-	httpclient := &http.Client{Transport: NewDeadlineTransport(2 * time.Second)}
+func apiRequest(endpoint string) (*simplejson.Json, error) {
+	httpclient := &http.Client{Transport: newDeadlineTransport(2 * time.Second)}
 	req, err := http.NewRequest("GET", endpoint, nil)
 	if err != nil {
 		return nil, err

command.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"strconv"
+	"time"
 )
 
 var byteSpace = []byte(" ")
@@ -27,44 +28,56 @@ func (c *Command) String() string {
 	return string(c.Name)
 }
 
-// Write serializes the Command to the supplied Writer.
+// WriteTo implements the WriterTo interface and
+// serializes the Command to the supplied Writer.
 //
-// It is suggested that the target Writer is buffered to avoid performing many system calls.
-func (c *Command) Write(w io.Writer) error {
-	_, err := w.Write(c.Name)
+// It is suggested that the target Writer is buffered
+// to avoid performing many system calls.
+func (c *Command) WriteTo(w io.Writer) (int64, error) {
+	var total int64
+	var buf [4]byte
+
+	n, err := w.Write(c.Name)
+	total += int64(n)
 	if err != nil {
-		return err
+		return total, err
 	}
 
 	for _, param := range c.Params {
-		_, err := w.Write(byteSpace)
+		n, err := w.Write(byteSpace)
+		total += int64(n)
 		if err != nil {
-			return err
+			return total, err
 		}
-		_, err = w.Write(param)
+		n, err = w.Write(param)
+		total += int64(n)
 		if err != nil {
-			return err
+			return total, err
 		}
 	}
 
-	_, err = w.Write(byteNewLine)
+	n, err = w.Write(byteNewLine)
+	total += int64(n)
 	if err != nil {
-		return err
+		return total, err
 	}
 
 	if c.Body != nil {
-		bodySize := int32(len(c.Body))
-		err := binary.Write(w, binary.BigEndian, &bodySize)
+		bufs := buf[:]
+		binary.BigEndian.PutUint32(bufs, uint32(len(c.Body)))
+		n, err := w.Write(bufs)
+		total += int64(n)
 		if err != nil {
-			return err
+			return total, err
 		}
-		_, err = w.Write(c.Body)
+		n, err = w.Write(c.Body)
+		total += int64(n)
 		if err != nil {
-			return err
+			return total, err
 		}
 	}
 
-	return nil
+	return total, nil
 }
 
 // Identify creates a new Command to provide information about the client.  After connecting,
@@ -92,7 +105,7 @@ func Register(topic string, channel string) *Command {
 	return &Command{[]byte("REGISTER"), params, nil}
 }
 
-// Unregister creates a new Command to remove a topic/channel for the connected nsqd
+// UnRegister creates a new Command to remove a topic/channel for the connected nsqd
 func UnRegister(topic string, channel string) *Command {
 	params := [][]byte{[]byte(topic)}
 	if len(channel) > 0 {
@@ -165,10 +178,10 @@ func Finish(id MessageID) *Command {
 }
 
 // Requeue creates a new Command to indicate that
-// a given message (by id) should be requeued after the given timeout (in ms)
-// NOTE: a timeout of 0 indicates immediate requeue
-func Requeue(id MessageID, timeoutMs int) *Command {
-	var params = [][]byte{id[:], []byte(strconv.Itoa(timeoutMs))}
+// a given message (by id) should be requeued after the given delay
+// NOTE: a delay of 0 indicates immediate requeue
+func Requeue(id MessageID, delay time.Duration) *Command {
+	var params = [][]byte{id[:], []byte(strconv.Itoa(int(delay / time.Millisecond)))}
 	return &Command{[]byte("REQ"), params, nil}
 }
 

command_test.go

@@ -18,6 +18,6 @@ func BenchmarkCommand(b *testing.B) {
 	b.StartTimer()
 
 	for i := 0; i < b.N; i++ {
-		cmd.Write(&buf)
+		cmd.WriteTo(&buf)
 	}
 }