hermes

README

hermes

http and grpc server implementation using fasthttp and aimed at connect-go users.

Documentation

Index

• Constants
• func CallbackSetUserID(data interface{}, req *Request)
• func CreateTopicsIfNotExist(ca sarama.ClusterAdmin, cl sarama.Client, topics []string) error
• func DefaultSaramaConfig(username, password string) *sarama.Config
• func PathBase(path string) string
• func SetVerbosity(b bool)
• func TopicMismatch(cl sarama.Client, topics []string) (bool, error)
• type AuthenticatorCallback
• type AuthenticatorFunc
• type Consumer
  • func NewConsumer(cfg *ConsumerConfig, handler ConsumerHandler) (*Consumer, error)
  • func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (c *Consumer) Setup(sarama.ConsumerGroupSession) error
  • func (c *Consumer) StartListening(ctxx context.Context) error
  • func (c *Consumer) ToggleConsumption()
• type ConsumerConfig
  • func (c *ConsumerConfig) Sarama() *sarama.Config
• type ConsumerHandler
• type Event
  • func (e *Event) Encode() (string, error)
• type EventType
  • func GetEventType(value []byte) (EventType, error)
• type Filter
  • func FilterAuthenticate(a AuthenticatorFunc, c AuthenticatorCallback) Filter
  • func FilterAuthenticateAndSetUserID(a AuthenticatorFunc) Filter
• type KafkaContextKey
• type Producer
  • func NewProducer(cfg *ProducerConfig) (*Producer, error)
  • func (p *Producer) Messages() chan *sarama.ProducerMessage
  • func (p *Producer) StartListening() error
• type ProducerConfig
  • func (c *ProducerConfig) Sarama() *sarama.Config
• type Request
  • func (r *Request) Ctx() context.Context
  • func (r *Request) Err() error
  • func (r *Request) Respond(v interface{}, code int) Response
  • func (r *Request) SetCtx(ctx context.Context)
  • func (r *Request) SetErr(err error)
• type Response
  • func FilterAttachTraceID(svc Service, req *Request) Response
  • func NewResponse(req *Request, b interface{}) Response
  • func NewResponseWithCode(req *Request, b interface{}, code int) Response
  • func (r *Response) Decode(v interface{}) error
  • func (r *Response) Encode(v interface{})
  • func (r *Response) FromConnect(v *connect.Request[any])
  • func (r *Response) JSON(v interface{})
  • func (r *Response) ToConnect(v *connect.Request[any]) error
• type Router
  • func (r *Router) CONNECT(pattern string, svc Service)
  • func (r *Router) DELETE(pattern string, svc Service)
  • func (r *Router) GET(pattern string, svc Service)
  • func (r *Router) Group(pattern string) *RouterGroup
  • func (r *Router) HEAD(pattern string, svc Service)
  • func (r Router) Lookup(method, path string) (Service, string, map[string]string, bool)
  • func (r *Router) OPTIONS(pattern string, svc Service)
  • func (r *Router) PATCH(pattern string, svc Service)
  • func (r *Router) POST(pattern string, svc Service)
  • func (r *Router) PUT(pattern string, svc Service)
  • func (r Router) Params(req Request) map[string]string
  • func (r Router) Pattern(req Request) string
  • func (r *Router) Register(method, pattern string, svc Service)
  • func (r Router) Serve() Service
  • func (r *Router) TRACE(pattern string, svc Service)
• type RouterGroup
  • func (r *RouterGroup) CONNECT(pattern string, svc Service)
  • func (r *RouterGroup) DELETE(pattern string, svc Service)
  • func (r *RouterGroup) GET(pattern string, svc Service)
  • func (r *RouterGroup) HEAD(pattern string, svc Service)
  • func (r *RouterGroup) OPTIONS(pattern string, svc Service)
  • func (r *RouterGroup) PATCH(pattern string, svc Service)
  • func (r *RouterGroup) POST(pattern string, svc Service)
  • func (r *RouterGroup) PUT(pattern string, svc Service)
  • func (r *RouterGroup) Register(method, pattern string, svc Service)
  • func (r *RouterGroup) TRACE(pattern string, svc Service)
• type ServeOptions
  • func DefaultServeOptions() *ServeOptions
• type Server
  • func NewServer() *Server
  • func (s *Server) RegisterRPC(path string, handler http.Handler)
  • func (s *Server) RegisterShutdownFunc(fs ...func(context.Context))
  • func (s *Server) Serve(svc Service, opt *ServeOptions)
  • func (s *Server) Stop(ctx context.Context)
• type Service
  • func (s Service) Filter(f Filter) Service

Constants

const (
	KafkaClusterVersion = "2.1.1"
)

Functions

func CallbackSetUserID

func CallbackSetUserID(data interface{}, req *Request)

CallbackSetUserID is a generic callback func that assumes the data provided is a string, and sets that string to a UserID header.

func CreateTopicsIfNotExist

func CreateTopicsIfNotExist(ca sarama.ClusterAdmin, cl sarama.Client, topics []string) error

func DefaultSaramaConfig

func DefaultSaramaConfig(username, password string) *sarama.Config

func PathBase

func PathBase(path string) string

func SetVerbosity

func SetVerbosity(b bool)

func TopicMismatch

func TopicMismatch(cl sarama.Client, topics []string) (bool, error)

Types

type AuthenticatorCallback

type AuthenticatorCallback func(data interface{}, req *Request)

type AuthenticatorFunc

type AuthenticatorFunc func(ctx context.Context, req *Request) (interface{}, error)

type Consumer

type Consumer struct {
	Configuration *ConsumerConfig
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(cfg *ConsumerConfig, handler ConsumerHandler) (*Consumer, error)

func (*Consumer) Cleanup

func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error

func (*Consumer) ConsumeClaim

func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

func (*Consumer) Setup

func (c *Consumer) Setup(sarama.ConsumerGroupSession) error

func (*Consumer) StartListening

func (c *Consumer) StartListening(ctxx context.Context) error

func (*Consumer) ToggleConsumption

func (c *Consumer) ToggleConsumption()

type ConsumerConfig

type ConsumerConfig struct {
	Username            string
	Password            string
	Brokers             []string
	Topics              []string
	Group               string
	Assignor            string
	InitialOffsetOldest bool
	// contains filtered or unexported fields
}

func (*ConsumerConfig) Sarama

func (c *ConsumerConfig) Sarama() *sarama.Config

type ConsumerHandler

type ConsumerHandler func(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

type Event

type Event struct {
	Type    EventType   `json:"type"`
	Message interface{} `json:"message"`
}

func (*Event) Encode

func (e *Event) Encode() (string, error)

type EventType

type EventType int

func GetEventType

func GetEventType(value []byte) (EventType, error)

type Filter

type Filter func(svc Service, req *Request) Response

func FilterAuthenticate

func FilterAuthenticate(a AuthenticatorFunc, c AuthenticatorCallback) Filter

FilterAuthenticateAndSetUserData returns a filter function that takes in an authenticator func and a callback func. The authenticator should use the authenticate the request, and either return nil interface to return a 403, or an error to return a 500.

func FilterAuthenticateAndSetUserID

func FilterAuthenticateAndSetUserID(a AuthenticatorFunc) Filter

type KafkaContextKey

type KafkaContextKey struct{}

type Producer

type Producer struct {
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(cfg *ProducerConfig) (*Producer, error)

func (*Producer) Messages

func (p *Producer) Messages() chan *sarama.ProducerMessage

func (*Producer) StartListening

func (p *Producer) StartListening() error

type ProducerConfig

type ProducerConfig struct {
	Username string
	Password string
	Brokers  []string
	// contains filtered or unexported fields
}

func (*ProducerConfig) Sarama

func (c *ProducerConfig) Sarama() *sarama.Config

type Request

type Request struct {
	fasthttp.Request
	// contains filtered or unexported fields
}

func (*Request) Ctx

func (r *Request) Ctx() context.Context

func (*Request) Err

func (r *Request) Err() error

func (*Request) Respond

func (r *Request) Respond(v interface{}, code int) Response

func (*Request) SetCtx

func (r *Request) SetCtx(ctx context.Context)

func (*Request) SetErr

func (r *Request) SetErr(err error)

type Response

type Response struct {
	fasthttp.Response

	Request *Request
	// contains filtered or unexported fields
}

func FilterAttachTraceID

func FilterAttachTraceID(svc Service, req *Request) Response

func NewResponse

func NewResponse(req *Request, b interface{}) Response

func NewResponseWithCode

func NewResponseWithCode(req *Request, b interface{}, code int) Response

func (*Response) Decode

func (r *Response) Decode(v interface{}) error

func (*Response) Encode

func (r *Response) Encode(v interface{})

func (*Response) FromConnect

func (r *Response) FromConnect(v *connect.Request[any])

func (*Response) JSON

func (r *Response) JSON(v interface{})

func (*Response) ToConnect

func (r *Response) ToConnect(v *connect.Request[any]) error

type Router

type Router struct {
	// contains filtered or unexported fields
}

func (*Router) CONNECT

func (r *Router) CONNECT(pattern string, svc Service)

func (*Router) DELETE

func (r *Router) DELETE(pattern string, svc Service)

func (*Router) GET

func (r *Router) GET(pattern string, svc Service)

func (*Router) Group

func (r *Router) Group(pattern string) *RouterGroup

func (*Router) HEAD

func (r *Router) HEAD(pattern string, svc Service)

func (Router) Lookup

func (r Router) Lookup(method, path string) (Service, string, map[string]string, bool)

func (*Router) OPTIONS

func (r *Router) OPTIONS(pattern string, svc Service)

func (*Router) PATCH

func (r *Router) PATCH(pattern string, svc Service)

func (*Router) POST

func (r *Router) POST(pattern string, svc Service)

func (*Router) PUT

func (r *Router) PUT(pattern string, svc Service)

func (Router) Params

func (r Router) Params(req Request) map[string]string

func (Router) Pattern

func (r Router) Pattern(req Request) string

func (*Router) Register

func (r *Router) Register(method, pattern string, svc Service)

func (Router) Serve

func (r Router) Serve() Service

func (*Router) TRACE

func (r *Router) TRACE(pattern string, svc Service)

type RouterGroup

type RouterGroup struct {
	Pattern string
	// contains filtered or unexported fields
}

func (*RouterGroup) CONNECT

func (r *RouterGroup) CONNECT(pattern string, svc Service)

func (*RouterGroup) DELETE

func (r *RouterGroup) DELETE(pattern string, svc Service)

func (*RouterGroup) GET

func (r *RouterGroup) GET(pattern string, svc Service)

func (*RouterGroup) HEAD

func (r *RouterGroup) HEAD(pattern string, svc Service)

func (*RouterGroup) OPTIONS

func (r *RouterGroup) OPTIONS(pattern string, svc Service)

func (*RouterGroup) PATCH

func (r *RouterGroup) PATCH(pattern string, svc Service)

func (*RouterGroup) POST

func (r *RouterGroup) POST(pattern string, svc Service)

func (*RouterGroup) PUT

func (r *RouterGroup) PUT(pattern string, svc Service)

func (*RouterGroup) Register

func (r *RouterGroup) Register(method, pattern string, svc Service)

func (*RouterGroup) TRACE

func (r *RouterGroup) TRACE(pattern string, svc Service)

type ServeOptions

type ServeOptions struct {
	// contains filtered or unexported fields
}

func DefaultServeOptions

func DefaultServeOptions() *ServeOptions

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer() *Server

func (*Server) RegisterRPC

func (s *Server) RegisterRPC(path string, handler http.Handler)

func (*Server) RegisterShutdownFunc

func (s *Server) RegisterShutdownFunc(fs ...func(context.Context))

func (*Server) Serve

func (s *Server) Serve(svc Service, opt *ServeOptions)

Serves both HTTP and gRPC

func (*Server) Stop

func (s *Server) Stop(ctx context.Context)

type Service

type Service func(req *Request) Response

func (Service) Filter

func (s Service) Filter(f Filter) Service