README
¶
StormRPC ⚡
StormRPC is an abstraction or wrapper on NATS Request/Reply messaging capabilities.
It provides some convenient features including:
-
Middleware
Middleware are decorators around
HandlerFuncs. Some middleware are available within the package includingRequestID,Tracing(via OpenTelemetry)LoggerandRecoverer. -
Body encoding and decoding
Marshalling and unmarshalling request bodies to structs. JSON, Protobuf, and Msgpack are supported out of the box.
-
Deadline propagation
Request deadlines are propagated from client to server so both ends will stop processing once the deadline has passed.
-
Error propagation
Responses have an
Errorattribute and these are propagated across the wire without needing to tweak your request/response schemas.
Installation
Runtime Library
The runtime library package github.com/actatum/stormrpc contains common types like stormrpc.Error, stormrpc.Client and stormrpc.Server. If you aren't generating servers and clients from protobuf definitions you only need to import the stormrpc package.
$ go get github.com/actatum/stormrpc
Code Generator
You need to install go and the protoc compiler on your system. Then, install the protoc plugins protoc-gen-stormrpc and protoc-gen-go to generate Go code.
$ go install github.com/actatum/stormrpc/protoc-gen-stormrpc
$ go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28
Code generation examples can be found here
Basic Usage
Server
package main
import (
"context"
"log"
"os"
"os/signal"
"syscall"
"time"
"github.com/actatum/stormrpc"
"github.com/nats-io/nats.go"
)
func echo(ctx context.Context, req stormrpc.Request) stormrpc.Response {
var b any
if err := req.Decode(&b); err != nil {
return stormrpc.NewErrorResponse(req.Reply, err)
}
resp, err := stormrpc.NewResponse(req.Reply, b)
if err != nil {
return stormrpc.NewErrorResponse(req.Reply, err)
}
return resp
}
func main() {
srv, err := stormrpc.NewServer("echo", nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
srv.Handle("echo", echo)
go func() {
_ = srv.Run()
}()
log.Printf("👋 Listening on %v", srv.Subjects())
done := make(chan os.Signal, 1)
signal.Notify(done, syscall.SIGINT, syscall.SIGTERM)
<-done
log.Printf("💀 Shutting down")
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err = srv.Shutdown(ctx); err != nil {
log.Fatal(err)
}
}
Client
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/actatum/stormrpc"
"github.com/nats-io/nats.go"
)
func main() {
client, err := stormrpc.NewClient(nats.DefaultURL)
if err != nil {
log.Fatal(err)
}
defer client.Close()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
r, err := stormrpc.NewRequest("echo", map[string]string{"hello": "me"})
if err != nil {
log.Fatal(err)
}
resp := client.Do(ctx, r)
if resp.Err != nil {
log.Fatal(resp.Err)
}
fmt.Println(resp.Header)
var result map[string]string
if err = resp.Decode(&result); err != nil {
log.Fatal(err)
}
fmt.Printf("Result: %v\n", result)
}
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func MessageFromErr ¶
MessageFromErr retrieves the message from a given error. If the error is not of type Error, "unknown error" is returned.
Types ¶
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client represents a stormRPC client. It contains all functionality for making RPC requests to stormRPC servers.
type ClientOption ¶
type ClientOption interface {
// contains filtered or unexported methods
}
ClientOption represents functional options for configuring a stormRPC Client.
type ErrorCode ¶
type ErrorCode int
ErrorCode represents an enum type for stormRPC error codes.
const ( ErrorCodeUnknown ErrorCode = 0 ErrorCodeInternal ErrorCode = 1 ErrorCodeNotFound ErrorCode = 2 )
RPC ErrorCodes.
func CodeFromErr ¶
CodeFromErr retrieves the ErrorCode from a given error. If the error is not of type Error, ErrorCodeUnknown is returned.
type ErrorHandler ¶
ErrorHandler is the function signature for handling server errors.
type HandlerFunc ¶
HandlerFunc is the function signature for handling of a single request to a stormRPC server.
type Middleware ¶
type Middleware func(next HandlerFunc) HandlerFunc
Middleware is the function signature for wrapping HandlerFunc's to extend their functionality.
type Request ¶
type Request struct {
*nats.Msg
}
Request is stormRPC's wrapper around a nats.Msg and is used by both clients and servers.
func NewRequest ¶
func NewRequest(subject string, body any, opts ...RequestOption) (Request, error)
NewRequest constructs a new request with the given parameters. It also handles encoding the request body.
type RequestOption ¶
type RequestOption interface {
// contains filtered or unexported methods
}
RequestOption represents functional options for configuring a request.
func WithEncodeMsgpack ¶
func WithEncodeMsgpack() RequestOption
WithEncodeMsgpack is a RequestOption to encode the request body using the msgpack.Marshal method.
func WithEncodeProto ¶
func WithEncodeProto() RequestOption
WithEncodeProto is a RequestOption to encode the request body using the proto.Marshal method.
type Response ¶
type Response struct {
*nats.Msg
Err error
}
Response is stormRPC's wrapper around a nats.Msg and is used by both clients and servers.
func NewErrorResponse ¶
NewErrorResponse constructs a new error response with the given parameters.
func NewResponse ¶
func NewResponse(reply string, body any, opts ...ResponseOption) (Response, error)
NewResponse constructs a new response with the given parameters. It also handles encoding the response body.
type ResponseOption ¶
type ResponseOption RequestOption
ResponseOption represents functional options for configuring a response.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server represents a stormRPC server. It contains all functionality for handling RPC requests.
func NewServer ¶
func NewServer(name, natsURL string, opts ...ServerOption) (*Server, error)
NewServer returns a new instance of a Server.
func (*Server) Handle ¶
func (s *Server) Handle(subject string, fn HandlerFunc)
Handle registers a new HandlerFunc on the server.
func (*Server) Use ¶
func (s *Server) Use(mw ...Middleware)
Use applies all given middleware globally across all handlers.
type ServerOption ¶
type ServerOption interface {
// contains filtered or unexported methods
}
ServerOption represents functional options for configuring a stormRPC Server.
func WithErrorHandler ¶
func WithErrorHandler(fn ErrorHandler) ServerOption
WithErrorHandler is a ServerOption that allows for registering a function for handling server errors.
Source Files
Directories