NATS Micro Handlers

2024-03-18

The Synadia team has a nice package for creating micro services. In the Go client, this is the micro package.

The beautiful thing about this setup is it follows the same pattern as the std library HTTP handlers. The micro package has an interface defined as Handler(micro.Request)Handler(micro.Request). Functions can be wrapped with micro.HandlerFunc()micro.HandlerFunc() just like you can wrap an HTTP handler with http.HandlerFunc()http.HandlerFunc().

This allows for the same midleware layouts as you normally have in HTTP servers.

Errors

Handlers accept a micro.Requestmicro.Request which contains methods to retrieve the data on the request. A request also contains an Error(code, description string, data []byte, opts ...RespondOpt) errorError(code, description string, data []byte, opts ...RespondOpt) error method. Similar to http.Error()http.Error() this will return an error to the caller.

Returning errors like this however can be burdensome depending on the number of handlers and number of possible errors your application can return. Luckily, since the NATS team has put a lot of thought into this design, we can wrap handlers and return errors easily.

Error Handlers

Let’s implement a custom error handler that we can just return errors to and have it handle them based on error type.


ype HandlerWithErrors func(micro.Request) error
type ClientError struct {
        Code    int
        Details string
}
func (c ClientError) Error() string {
        return c.Details
}
func (c *ClientError) Body() []byte {
        return []byte(fmt.Sprintf(`{"error": "%s"}`, c.Details))
}
func (c *ClientError) CodeString() string {
        return strconv.Itoa(c.Code)
}
func (c ClientError) As(target any) bool {
        _, ok := target.(*ClientError)
        return ok
}
func NewClientError(err error, code int) ClientError {
        return ClientError{
                Code:    code,
                Details: err.Error(),
        }
}


ype HandlerWithErrors func(micro.Request) error
type ClientError struct {
        Code    int
        Details string
}
func (c ClientError) Error() string {
        return c.Details
}
func (c *ClientError) Body() []byte {
        return []byte(fmt.Sprintf(`{"error": "%s"}`, c.Details))
}
func (c *ClientError) CodeString() string {
        return strconv.Itoa(c.Code)
}
func (c ClientError) As(target any) bool {
        _, ok := target.(*ClientError)
        return ok
}
func NewClientError(err error, code int) ClientError {
        return ClientError{
                Code:    code,
                Details: err.Error(),
        }
}

Here we define a few things. A function type which we can use to define our specific handler type. A struct called ClientErrorClientError which will denote client errors. Some methods on ClientErrorClientError, one being Error() stringError() string which satisfies the error interface. And finally a function to return a new ClientError.

Middleware


func ErrorHandler(h HandlerWithErrors) micro.HandlerFunc {
        return func(r micro.Request) {
                err = h(r)
                if err == nil {
                        return
                }
          var ce ClientError
          if errors.As(err, &ce) {
                r.Error(ce.CodeString(), http.StatusText(ce.Code), ce.Body())
                return
          }
          logger.Error(err)
          r.Error("500", "internal server error", []byte(`{"error": "internal server error"}`))
        }
}


func ErrorHandler(h HandlerWithErrors) micro.HandlerFunc {
        return func(r micro.Request) {
                err = h(r)
                if err == nil {
                        return
                }
          var ce ClientError
          if errors.As(err, &ce) {
                r.Error(ce.CodeString(), http.StatusText(ce.Code), ce.Body())
                return
          }
          logger.Error(err)
          r.Error("500", "internal server error", []byte(`{"error": "internal server error"}`))
        }
}

Here we define the function that we can wrap our handlers with. Here we return a micro.HandlerFuncmicro.HandlerFunc which satisfies the micro.Handlermicro.Handler interface. In our code we can then just call svc.AddEndpoint("test", ErrorHandler(myHandler))svc.AddEndpoint("test", ErrorHandler(myHandler))

In our handlers, we simply need to return the error. The middleware will check the error type and handle it appropriately.

Sample Code

Let’s take a look at a sample program to do all of this. Start the service with the code below and send some sample requests.

nats req "test" '{"name": "Pete"}'nats req "test" '{"name": "Pete"}': This should return a 500 error

nats req "test" '{"name": "John"'nats req "test" '{"name": "John"': This should also return a 400 error

nats req "test" '{"name": "John"}'nats req "test" '{"name": "John"}': This should simply return “John”


package main
import (
        "encoding/json"
        "errors"
        "fmt"
        "log"
        "net/http"
        "os"
        "os/signal"
        "strconv"
        "syscall"
        "github.com/nats-io/nats.go"
        "github.com/nats-io/nats.go/micro"
)
func main() {
        nc, err := nats.Connect(nats.DefaultURL)
        if err != nil {
                log.Fatal(err)
        }
        ms, err := micro.AddService(nc, micro.Config{
                Name:        "test",
                Version:     "0.0.1",
                Description: "Example",
        })
        if err != nil {
                log.Fatal(err)
        }
        ms.AddEndpoint("test", ErrorHandler(myHandler))
        sigTerm := make(chan os.Signal, 1)
        signal.Notify(sigTerm, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
        sig := <-sigTerm
        log.Printf("received signal: %s\n", sig)
        ms.Stop()
}
type Person struct {
        Name string `json:"name"`
}
func myHandler(r micro.Request) error {
        var person Person
        if !json.Valid(r.Data()) {
                return NewClientError(fmt.Errorf("invalid JSON"), 400)
        }
        if err := json.Unmarshal(r.Data(), &person); err != nil {
                return NewClientError(err, 400)
        }
        if person.Name == "Pete" {
                return fmt.Errorf("this is a server error")
        }
        r.Respond(r.Data())
        return nil
}
type HandlerWithErrors func(micro.Request) error
type ClientError struct {
        Code    int
        Details string
}
func (c ClientError) Error() string {
        return c.Details
}
func (c *ClientError) Body() []byte {
        return []byte(fmt.Sprintf(`{"error": "%s"}`, c.Details))
}
func (c *ClientError) CodeString() string {
        return strconv.Itoa(c.Code)
}
func (c ClientError) As(target any) bool {
        _, ok := target.(*ClientError)
        return ok
}
func NewClientError(err error, code int) ClientError {
        return ClientError{
                Code:    code,
                Details: err.Error(),
        }
}
func ErrorHandler(h HandlerWithErrors) micro.HandlerFunc {
        return func(r micro.Request) {
                err := h(r)
                if err == nil {
                        return
                }
                var ce ClientError
                if errors.As(err, &ce) {
                        r.Error(ce.CodeString(), http.StatusText(ce.Code), ce.Body())
                        return
                }
                log.Println(err)
                r.Error("500", "internal server error", []byte(`{"error": "internal server error"}`))
        }
}


package main
import (
        "encoding/json"
        "errors"
        "fmt"
        "log"
        "net/http"
        "os"
        "os/signal"
        "strconv"
        "syscall"
        "github.com/nats-io/nats.go"
        "github.com/nats-io/nats.go/micro"
)
func main() {
        nc, err := nats.Connect(nats.DefaultURL)
        if err != nil {
                log.Fatal(err)
        }
        ms, err := micro.AddService(nc, micro.Config{
                Name:        "test",
                Version:     "0.0.1",
                Description: "Example",
        })
        if err != nil {
                log.Fatal(err)
        }
        ms.AddEndpoint("test", ErrorHandler(myHandler))
        sigTerm := make(chan os.Signal, 1)
        signal.Notify(sigTerm, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
        sig := <-sigTerm
        log.Printf("received signal: %s\n", sig)
        ms.Stop()
}
type Person struct {
        Name string `json:"name"`
}
func myHandler(r micro.Request) error {
        var person Person
        if !json.Valid(r.Data()) {
                return NewClientError(fmt.Errorf("invalid JSON"), 400)
        }
        if err := json.Unmarshal(r.Data(), &person); err != nil {
                return NewClientError(err, 400)
        }
        if person.Name == "Pete" {
                return fmt.Errorf("this is a server error")
        }
        r.Respond(r.Data())
        return nil
}
type HandlerWithErrors func(micro.Request) error
type ClientError struct {
        Code    int
        Details string
}
func (c ClientError) Error() string {
        return c.Details
}
func (c *ClientError) Body() []byte {
        return []byte(fmt.Sprintf(`{"error": "%s"}`, c.Details))
}
func (c *ClientError) CodeString() string {
        return strconv.Itoa(c.Code)
}
func (c ClientError) As(target any) bool {
        _, ok := target.(*ClientError)
        return ok
}
func NewClientError(err error, code int) ClientError {
        return ClientError{
                Code:    code,
                Details: err.Error(),
        }
}
func ErrorHandler(h HandlerWithErrors) micro.HandlerFunc {
        return func(r micro.Request) {
                err := h(r)
                if err == nil {
                        return
                }
                var ce ClientError
                if errors.As(err, &ce) {
                        r.Error(ce.CodeString(), http.StatusText(ce.Code), ce.Body())
                        return
                }
                log.Println(err)
                r.Error("500", "internal server error", []byte(`{"error": "internal server error"}`))
        }
}