README
¶
Requirements
The engine stores its state in PostgreSQL. The minimum supported version is PostgreSQL 18.
Documentation
¶
Overview ¶
Package runkit is a reference implementation of a Dogma engine.
Index ¶
Constants ¶
const ( // DefaultProjectionCompactInterval is the default minimum time between // successful projection compactions. DefaultProjectionCompactInterval = 6 * time.Hour )
Variables ¶
This section is empty.
Functions ¶
func CreateSchema ¶ added in v0.1.2
CreateSchema creates the database schema used by the engine.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine is a Dogma engine backed by a single PostgreSQL database.
It implements dogma.CommandExecutor.
It must be created with New; the zero value is not usable.
func New ¶ added in v0.2.0
func New( app dogma.Application, db *sql.DB, options ...EngineOption, ) *Engine
New returns a new engine that runs the given application.
db must use the pgx driver, for example by opening it with sql.Open("pgx", dsn) or stdlib.OpenDB. It panics if db is nil or uses any other driver.
func (*Engine) ExecuteCommand ¶
func (e *Engine) ExecuteCommand( ctx context.Context, command dogma.Command, options ...dogma.ExecuteCommandOption, ) error
ExecuteCommand submits a [Command] for execution.
It returns once the engine has taken ownership of the command. By default, it doesn't wait for handling to finish.
See dogma.CommandExecutor for more details.
func (*Engine) ListenAddr ¶ added in v0.2.0
ListenAddr returns the address that the engine is listening on for gRPC requests from other engines.
It returns an error immediately if the engine was created without WithListenAddress. Otherwise, it blocks until the engine is ready. It returns an error if ctx is canceled before then, or if the engine has stopped.
func (*Engine) Ready ¶
func (e *Engine) Ready() <-chan struct{}
Ready returns a channel that is closed when the engine is ready to accept commands for execution.
It is exposed as an operational signal. For example, to signal readiness to a load balancer. It is not an error to attempt command execution before the engine is ready.
type EngineOption ¶ added in v0.2.0
type EngineOption func(*Engine)
EngineOption is a function that configures an Engine.
func WithListenAddress ¶ added in v0.2.0
func WithListenAddress(addr string, options ...grpc.ServerOption) EngineOption
WithListenAddress is an EngineOption that makes the engine listen on addr for gRPC requests from other engines, such as requests to consume the application's events.
If it is not provided the engine does not listen for gRPC requests. It panics if addr is empty.
The options are used to create the gRPC server. As with any gRPC server, it is insecure by default: unless transport credentials are supplied using grpc.Creds, connections are neither encrypted nor authenticated. Any client that is able to connect can consume all of the application's events, so the credentials should authenticate clients, for example by using mutual TLS:
runkit.WithListenAddress(
":50555",
grpc.Creds(
credentials.NewTLS(&tls.Config{
Certificates: []tls.Certificate{serverCert},
ClientCAs: clientCAs,
ClientAuth: tls.RequireAndVerifyClientCert,
}),
),
)
func WithLogger ¶ added in v0.2.0
func WithLogger(logger *slog.Logger) EngineOption
WithLogger is an EngineOption that sets the logger the engine uses.
If it is not provided slog.Default is used.
func WithProjectionCompactInterval ¶ added in v0.2.0
func WithProjectionCompactInterval(interval time.Duration) EngineOption
WithProjectionCompactInterval is an EngineOption that sets the minimum time between successful projection compactions.
If it is not provided DefaultProjectionCompactInterval is used. It panics if interval is not positive.
func WithRemoteEngineDiscoverer ¶ added in v0.2.0
func WithRemoteEngineDiscoverer(d RemoteEngineDiscoverer) EngineOption
WithRemoteEngineDiscoverer is an EngineOption that adds d to the discoverers that the engine uses to find remote engines.
It may be provided more than once, adding a discoverer each time. Use StaticRemoteEngines for a fixed set of remote engines. It panics if d is nil.
type RemoteEngineDiscoverer ¶ added in v0.2.0
type RemoteEngineDiscoverer func( ctx context.Context, add func(conn grpc.ClientConnInterface) (remove func()), ) error
RemoteEngineDiscoverer discovers remote engines, which are other engines that this engine communicates with, such as to consume their events.
A remote engine is any endpoint that serves Dogma gRPC services. For engines that have multiple replicas, it represents the logical engine as a whole, rather than any individual replica.
It calls the add function for each remote engine as it becomes available, passing a connection to the remote engine, and calls the remove function that it returns when the remote engine is no longer available. The add function panics if the connection is nil.
This engine uses the connection until the remove function is called, but never closes it. The remove function blocks until the engine has stopped using the connection, after which it may be closed. Calling the remove function more than once has no effect.
If two remote engines offer the same resource, such as the same event stream, this engine chooses one of them to use. There are no guarantees about which is chosen, and the choice may change over time even if the set of available remote engines remains the same.
The discoverer should run until ctx is canceled. If it returns before then, every remote engine that it added is removed and it is called again after a delay. The add function must not be called after the discoverer returns.
func StaticRemoteEngines ¶ added in v0.2.0
func StaticRemoteEngines(conns ...grpc.ClientConnInterface) RemoteEngineDiscoverer
StaticRemoteEngines returns a RemoteEngineDiscoverer that adds each of the given connections, and never removes them.
It panics if any of the connections is nil.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
aggregate
Package aggregate provides engine components that orchestrate the execution of aggregate message handlers.
|
Package aggregate provides engine components that orchestrate the execution of aggregate message handlers. |
|
application
Package application binds the database to a single Dogma application.
|
Package application binds the database to a single Dogma application. |
|
concurrency
Package concurrency provides utilities for enforcing a handler's concurrency preference.
|
Package concurrency provides utilities for enforcing a handler's concurrency preference. |
|
contexthook
Package contexthook provides utilities for adding request-scoped testing hooks to contexts.
|
Package contexthook provides utilities for adding request-scoped testing hooks to contexts. |
|
eventstream
Package eventstream provides a catalog of the event streams that the engine consumes, both this application's own and those of remote engines.
|
Package eventstream provides a catalog of the event streams that the engine consumes, both this application's own and those of remote engines. |
|
eventstream/internal/cache
Package cache provides a cache of an event stream's most recent events.
|
Package cache provides a cache of an event stream's most recent events. |
|
eventstream/source
Package source defines where an event stream's events come from, the contract by which a source delivers them, and the gRPC service that provides a source's event streams to remote engines.
|
Package source defines where an event stream's events come from, the contract by which a source delivers them, and the gRPC service that provides a source's event streams to remote engines. |
|
grpcserver
Package grpcserver serves the engine's gRPC services to other engines.
|
Package grpcserver serves the engine's gRPC services to other engines. |
|
integration
Package integration provides engine components that orchestrate the execution of integration message handlers.
|
Package integration provides engine components that orchestrate the execution of integration message handlers. |
|
messagepump
Package messagepump generalizes a message pump that acquires messages for processing and dispatches them to a worker pool.
|
Package messagepump generalizes a message pump that acquires messages for processing and dispatches them to a worker pool. |
|
process
Package process provides engine components that orchestrate the execution of process message handlers.
|
Package process provides engine components that orchestrate the execution of process message handlers. |
|
projection
Package projection provides engine components that orchestrate the execution of projection message handlers.
|
Package projection provides engine components that orchestrate the execution of projection message handlers. |
|
remoteengine
Package remoteengine provides engine components that connect the engine to remote engines.
|
Package remoteengine provides engine components that connect the engine to remote engines. |
|
schema
Package schema defines the PostgreSQL schema used by the engine.
|
Package schema defines the PostgreSQL schema used by the engine. |
|
signal
Package signal provides a mechanism for waking components when something they are interested in changes, using the PostgreSQL LISTEN/NOTIFY feature.
|
Package signal provides a mechanism for waking components when something they are interested in changes, using the PostgreSQL LISTEN/NOTIFY feature. |
|
supervisor
Package supervisor runs fallible functions until they succeed, retrying with exponential backoff when they fail.
|
Package supervisor runs fallible functions until they succeed, retrying with exponential backoff when they fail. |
|
x/xerrors
Package xerrors provides utilities for error handling.
|
Package xerrors provides utilities for error handling. |
|
x/xsql
Package xsql provides utilities for working with SQL databases.
|
Package xsql provides utilities for working with SQL databases. |
|
x/xtesting
Package xtesting provides utilities for testing.
|
Package xtesting provides utilities for testing. |