crossbow

package module
v0.0.0-...-a19bde6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 18, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

README

Crossbow

Is a simple actor-like, inbox based worker-pool. It provides: lightweight lifecycle managment, panic recovery, adaptable mailbox policies and resizing, synchronous and asynchronous communications with the server; optional handler level parallelism and classic FIFO processing as default.

Crossbow makes heavy use of generics to keep whole thing type safe. It is kept lean and lightweight on purpose. That is why it lacks a central registry, managers or process trees.

Toolchain

Go 1.25 or later is required. The 'go.uber.org/goleak' package is used for tests built with -tags=leaktest, it is encouraged to run such test after making changes to the code base. This module requires 'go.uber.org/goleak' but, it is not neccesary for the library functions.

Why Crossbow

Crossbow offers actor-like stage managment in a lightweight and type-safe fashion; while at the same time extending it with an optional worker pool.

Crossbow offer configurable panic recovery that lets your provide your own handler to inspect the mssage causing the panic and the state of the handler.

Included with the library is sensible set of mailbox backpressure policies, though by providing a simple interface you can write your own backpressure handlers.

Crossbow allows you to limit the scope of every operation to an appropriate handler, making your code for modular and safer.

Versioning

Crossbow follows SemVer. Backward compatibility will be assured from v1 onwards. We intends to use the standard go support cycle, each version of crossbow will be maintained for two version of the Go language.

Usage

All examples can be found in ./examples

To use Crossbow you need to define a handler that implements the ServerHandler[M any, O any] interface. Most importantly you need to have a Handle(ContextMessage[M, O]) method, this is the handling loop of your server; any data sent ot it is popped from the inbox and passed to it.

(...)

// Simple stateless handler that send back data through the provided channel
type Foo struct{
    x int
}

func (e *Foo) Handle(msg ContextMessage[int, int]) (int, error) {
    if e.x == 2137 {
        fmt.Println("21:37")
        e.x = 0
        return 0, nil
    } else {
        e.x = msg.Value // Update internal state. Safe if the thread count is 1
        return msg.Value, nil
    }
}

// Simple init stub. Usually used to initialize connections and validate fields
func (e *Foo) Init() error {
	return nil
}

// Handles the closing of connections etc.
func (e *Foo) Terminate(err error) {
    if e.x == 42 {
        fmt.Println("Ah! 42")
    }
}

func main() {
    handler := &Foo{}
    cfg := MakeDefaultServerConfig(1) // one worker for sequential processing
    srv, _ := NewServer[*Foo, int, int](handler, cfg, DefaultPanicRecover)

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go srv.Run(ctx) // this must be done before sending anything to the server

    rqCtx, _ := context.WithCancel(ctx) // canceling a requests context won't stop the whole server
    _ = srv.Send(rqCtx, 2137) // Send() does not block awaiting a response and just discards any

    out, err := srv.Call(rqCtx, 11)
    // Output will be 0; err will be nil; and the string "21:37" will be printed
}

Documentation

Overview

Package crossbow Is a simple actor-like, inbox based worker-pool. It provides: lightweight lifecycle managment, panic recovery, adaptable mailbox policies and resizing, synchronous and asynchronous communications with the server; optional handler level parallelism and classic FIFO processing as default.

Crossbow makes heavy use of generics to keep whole thing type safe. It is kept lean and lightweight on purpose. That is why it lacks a central registry, managers or process trees.

Why Crossbow

Crossbow offers actor-like stage managment in a lightweight and type-safe fashion; while at the same time extending it with an optional worker pool.

Crossbow offer configurable panic recovery that lets your provide your own handler to inspect the mssage causing the panic and the state of the handler.

Included with the library is sensible set of mailbox backpressure policies, though by providing a simple interface you can write your own backpressure handlers.

Crossbow allows you to limit the scope of every operation to an appropriate handler, making your code for modular and safer.

Versioning

Crossbow is intended to strcitly follow SemVer from version v0.1 onwards. Releases marked as v0-alpha.x.y or v0-beta.x.y do not make any compatibility guarantees. We intends to use the standard go support cycle, each version of crossbow will be maintained for two version of the Go language.

Usage

All examples can be found in ./examples

To use Crossbow you need to define a handler that implements the ServerHandler[M any, O any] interface. Most importantly you need to have a Handle(ContextMessage[M, O]) method, this is the handling loop of your server; any data sent ot it is popped from the inbox and passed to it. 

	// Simple stateless handler that send back data through the provided channel
	type Foo struct{
	    x int
	}

	func (e *Foo) Handle(msg ContextMessage[int, int]) (int, error) {
	    if e.x == 2137 {
	        fmt.Println("21:37")
	        e.x = 0
	        return 0, nil
	    } else {
	        e.x = msg.Value // Update internal state. Safe if the thread count is 1
	        return msg.Value, nil
	    }
	}

	// Simple init stub. Usually used to initialize connections and validate fields
	func (e *Foo) Init() error {
		return nil
	}

	// Handles the closing of connections etc.
	func (e *Foo) Terminate(err error) {
	    if e.x == 42 {
	        fmt.Println("Ah! 42")
	    }
	}

	func main() {
		handler := &Foo{}
		cfg := MakeDefaultServerConfig(1) // one worker for sequential processing
		srv, _ := NewServer[*Foo, int, int](handler, cfg, DefaultPanicRecover)

		ctx, cancel := context.WithCancel(context.Background())
		defer cancel()
		go srv.Run(ctx) // this must be done before sending anything to the server

		rqCtx, _ := context.WithCancel(ctx) // canceling a requests context won't stop the whole server
		_ = srv.Send(rqCtx, 2137) // Send() does not block awaiting a response and just discards any

    	out, err := srv.Call(rqCtx, 11)
		// Output will be 0; err will be nil; and the string "21:37" will be printed
	}

Toolchain

Go 1.25 or later is required. The 'go.uber.org/goleak' package is used for tests built with `-tags=leaktest`, it is encouraged to run such test after making changes to the code base. This module requires 'go.uber.org/goleak' but, it is not neccesary for the library functions.

Index

Constants

This section is empty.

Variables

View Source 
var ErrServerTerminated = errors.New("server is terminated")

Error returned senders of messages inside the queue if the server is terminated before the message is handled

Functions

func DefaultPanicRecover 

func DefaultPanicRecover[T ServerHandler[M, O], M any, O any](msg ContextMessage[M, O], handler T, err error, stack []byte) error

DefaultPanicRecover is a basic handler for panics inside of the handler. It will return en error causing the server to terminate.

Types

type ContextMessage 

type ContextMessage[M any, O any] struct {
	Value M // The message being sent to the handler

	Context   context.Context // Context to be passed to handlers
	Timestamp time.Time       // Timestamp,  populated automatically if makeTimestamp is true
	// contains filtered or unexported fields
}

ContextMessage holds a message, a channel where a response should be sent and general context such as.

type CustomMailboxPolicy 

type CustomMailboxPolicy[T any] interface {
	queue.MailboxPolicy[T] // required function: ` Enqueue(context.Context, *Queue[T], T) error `
}

CustomMailboxPolicy defines rules for handling attempts to push to a full queue. The last argument is the item being pushed, the second one the the queue object crossbow/internal/queue. The first argument is the context passed by the caller, you do not need to handle it in any special way. Return an apropriate error if the pushing fails, if so, you generally should not modify the underlying queues content so as to not confuse callers.

type MailboxPolicy 

type MailboxPolicy int

MailboxPolicy represents one of the predefined MailboxPolicies provided by the library. 

const (
	PolicyBlock      MailboxPolicy = iota // Default policy, if the queue is full Send() or Call() will block until there is space in the queue
	PolicyUnbounded                       // The queue is allowed to grow indefnitely, generally dicouraged for most use cases
	PolicyDropNewest                      // If the queue if full Send() or Call() will drop the message and return an appropriate error
	PolicyDropOldest                      // If the queue if full Send() or Call() will drop the oldest message (first in line) and the enqueue the just passed message. If a failure occurs ErrFull will be returned
)

type RecoverFn 

type RecoverFn[T ServerHandler[M, O], M any, O any] = func(ContextMessage[M, O], T, error, []byte) error // function called when the handler panics, if it returns an erros the server terminates

type Response 

type Response[O any] struct {
	Value O     // Result of handling the message
	Err   error // Error returned from the handler
}

Response from a handler that carries the error. It is up to the programmers discrection when an error should be returned and if Value can be valid if Err != nil.

type Server 

type Server[T ServerHandler[M, O], M any, O any] struct {
	// contains filtered or unexported fields
}

Server is the central object for crossbow. It represents a handler with a inbox attached. The handler processes messages from the inbox in a FIFO manner by default. The first type parameter is a user defined handler that implements the ServerHandler interface. M is the message type that the handler acccepts wrapped with additional context by ContextMessage. O is the output from the handler, once the handler finishes it must return O and error, both will be passed on to the caller when using the Call method, and disacarded if using Send. The server is threadsafe within its public APIs. The default config, provided by MakeDefaultServerConfig MakeDefaultServerConfig, sets the max worker count to 1; this guarantees FIFO processing od messages sent like in regular actor systems.

func NewServer 

func NewServer[T ServerHandler[M, O], M any, O any](handler T, cfg ServerConfig, recover RecoverFn[T, M, O]) (*Server[T, M, O], error)

NewServer returns a ready to run server. Importantly, in this state the server may hang forever on any messages sent. Handling of deadlines and cancelations if only guaranteed after running Server.Run() [Server.Run()]. New Server accepts the user-defined handler and server config ServerConfig as its agruments. It run the Validate() [ServerConfig.Validate()] methods on the passed config, returning early with the errors and nil passed as the server pointer. NewServer is also responsible for running Init() on the handler. Upon encoutering an error it returns that error and nil as the server pointer. NewSerevr initializes the queue that back the mailbox with setting retrived from ServerConfig. It initializes the queue with a size of `1 + cfg.Worker`.

func (*Server[T, M, O]) Call 

func (s *Server[T, M, O]) Call(ctx context.Context, req M) (O, error)

Call is the basic method for talking to the handler, it blocks until the handler returns a response. It returns a context error upon cancelation of ctx. An error is returned if creating a new ContextMessage fail. An error is also returned if enqueueing the message fails. If nil is returned that means that the message was successfully enqueued

func (*Server[T, M, O]) MailboxLen 

func (s *Server[T, M, O]) MailboxLen() int

MailboxLen returns the length of the queue that backs the Server's mailbox

func (*Server[T, M, O]) Run 

func (s *Server[T, M, O]) Run(ctx context.Context)

Run sets up the server's main loop and defers cleanup functions. It must be run before sending anything to the server. It returns upon context cancelation or s.terminated being true or the queues.Notify() reporting the underlying channel to be closed. The main loop awaits new data and wakes up when the queue notifies it.

func (*Server[T, M, O]) Send 

func (s *Server[T, M, O]) Send(ctx context.Context, req M) error

Send is the basic method for signaling to the handler, it does not block waiting for a response. It returns a context error upon cancelation of ctx. An error is returned if creating a new ContextMessage fail. An error is also returned if enqueueing the message fails. If nil is returned that means that the message was successfully enqueued

func (*Server[T, M, O]) Stats 

func (s *Server[T, M, O]) Stats() StatsSnapshot

Stats returns a StatsSnapshot StatsSnapshot, it reads the atomic Uints in the dynamic stats and saves it into the structure. It is thus guranteed to be thread safe but, the stats may change from the time the values were saved and this function outputed the snapshot.

func (*Server[T, M, O]) Terminated 

func (s *Server[T, M, O]) Terminated() bool

Terminated reports whether the server has terminated.

type ServerConfig 

type ServerConfig struct {
	Workers            uint          // The amount of workers to run the handler over, 1 for standard FIFO ordering
	MailboxSize        uint          // Maximum size of a the servers mailbox, must be greater than 0
	Policy             MailboxPolicy // MailboxPolicy according to [MailboxPolicy]
	GenerateTimestamps bool          // Switch for autogeneration of timestamps for the Timestamp field on ContextMessage [ContextMessage]
}

ServerConfig provides a recovery function settings for a server and facilitates easy validation

func MakeDefaultServerConfig 

func MakeDefaultServerConfig(workers ...uint) ServerConfig

MakeDefaultServerConfig provides a sesible default configuration for a server. If no wokers argument is provided a default of 1 is used. Arguments past the first one are ignored, but the library reserves a right to use them for any purpose in the future will cause the server to terminate. It also uses the PolicyBlock for its mailbox. The mailbox size is set to (4 + Workers * 4).

func (*ServerConfig) Validate 

func (s *ServerConfig) Validate() error

Validate checks the fields of ServerConfig for disallowed values, returns an apropriate erros when a disallowed value is found.

type ServerConfigError 

type ServerConfigError struct {
	ErrType ServerConfigErrorType
	Msg     string
}

Error struct for ServerConfig validation errors; msg may be used to hold additional data about the error.

func (ServerConfigError) Error 

func (s ServerConfigError) Error() string

type ServerConfigErrorType 

type ServerConfigErrorType int

Possible error types when validating server configs ServerConfig. 

const (
	ConfigErrorWorkersZero     ServerConfigErrorType = iota // Workers cannot be zero, at least one thread must be dedicated to the handler
	ConfigErrorRecoverFuncNil                               // The panic recovery is not allowed to be nil
	ConfigErrorMailboxSizeZero                              // The mailbox must not be zero
)

func (ServerConfigErrorType) String 

func (i ServerConfigErrorType) String() string

type ServerHandler 

type ServerHandler[M any, O any] interface {
	Init() error                            // called at server startup
	Handle(ContextMessage[M, O]) (O, error) // Handle incoming messages, called repeatadly for that purpose
	Terminate(reason error)                 // called at server shutdown
}

type StatsSnapshot 

type StatsSnapshot struct {
	Failures uint64 // amount of failures since start
	Panics   uint64 // amount of panis since start
}

ServerStats as regular ints

Source Files

View all Source files

Directories

Path Synopsis
internal
queue
ring

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.