processing

Documentation

Overview

Package processing provides event processing services for the ORLY relay. It handles event persistence, delivery to subscribers, and post-save hooks.

Index

• type ACLRegistry
• type ClusterManager
• type Config
  • func DefaultConfig() *Config
• type Database
• type Publisher
• type RateLimiter
• type RelayGroupManager
• type Result
  • func Blocked(msg string) Result
  • func Failed(err error) Result
  • func OK() Result
• type Service
  • func New(cfg *Config, db Database, publisher Publisher) *Service
  • func (s *Service) Process(ctx context.Context, ev *event.E) Result
  • func (s *Service) SetACLRegistry(acl ACLRegistry)
  • func (s *Service) SetClusterManager(cm ClusterManager)
  • func (s *Service) SetRateLimiter(rl RateLimiter)
  • func (s *Service) SetRelayGroupManager(rgm RelayGroupManager)
  • func (s *Service) SetSyncManager(sm SyncManager)
• type SyncManager

Types

type ACLRegistry

type ACLRegistry interface {
	// Configure reconfigures the ACL system.
	Configure(cfg ...any) error
	// Active returns the active ACL mode.
	Active() string
}

ACLRegistry abstracts ACL registry for reconfiguration.

type ClusterManager

type ClusterManager interface {
	// HandleMembershipEvent processes a cluster membership event.
	HandleMembershipEvent(ev *event.E) error
}

ClusterManager handles cluster membership events.

type Config

type Config struct {
	Admins       [][]byte
	Owners       [][]byte
	WriteTimeout time.Duration
}

Config holds configuration for the processing service.

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns the default processing configuration.

type Database

type Database interface {
	// SaveEvent saves an event to the database.
	SaveEvent(ctx context.Context, ev *event.E) (exists bool, err error)
	// CheckForDeleted checks if an event has been deleted.
	CheckForDeleted(ev *event.E, adminOwners [][]byte) error
}

Database abstracts database operations for event processing.

type Publisher

type Publisher interface {
	// Deliver sends an event to all matching subscribers.
	Deliver(ev *event.E)
}

Publisher abstracts event delivery to subscribers.

type RateLimiter

type RateLimiter interface {
	// IsEnabled returns whether rate limiting is enabled.
	IsEnabled() bool
	// Wait blocks until the rate limit allows the operation.
	Wait(ctx context.Context, opType int) error
}

RateLimiter abstracts rate limiting for write operations.

type RelayGroupManager

type RelayGroupManager interface {
	// ValidateRelayGroupEvent validates a relay group config event.
	ValidateRelayGroupEvent(ev *event.E) error
	// HandleRelayGroupEvent processes a relay group event.
	HandleRelayGroupEvent(ev *event.E, syncMgr any)
}

RelayGroupManager handles relay group configuration events.

type Result

type Result struct {
	Saved     bool
	Duplicate bool
	Blocked   bool
	BlockMsg  string
	Error     error
}

Result contains the outcome of event processing.

func Blocked

func Blocked(msg string) Result

Blocked returns a blocked processing result.

func Failed

func Failed(err error) Result

Failed returns an error processing result.

func OK

func OK() Result

OK returns a successful processing result.

type Service

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

Service implements event processing.

func New

func New(cfg *Config, db Database, publisher Publisher) *Service

New creates a new processing service.

func (*Service) Process

func (s *Service) Process(ctx context.Context, ev *event.E) Result

Process saves an event and triggers delivery.

func (*Service) SetACLRegistry

func (s *Service) SetACLRegistry(acl ACLRegistry)

SetACLRegistry sets the ACL registry.

func (*Service) SetClusterManager

func (s *Service) SetClusterManager(cm ClusterManager)

SetClusterManager sets the cluster manager.

func (*Service) SetRateLimiter

func (s *Service) SetRateLimiter(rl RateLimiter)

SetRateLimiter sets the rate limiter.

func (*Service) SetRelayGroupManager

func (s *Service) SetRelayGroupManager(rgm RelayGroupManager)

SetRelayGroupManager sets the relay group manager.

func (*Service) SetSyncManager

func (s *Service) SetSyncManager(sm SyncManager)

SetSyncManager sets the sync manager.

type SyncManager

type SyncManager interface {
	// UpdateSerial updates the serial number after saving an event.
	UpdateSerial()
}

SyncManager abstracts sync manager for serial updates.