stream

package
v0.6.10 Latest Latest
Warning

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

Go to latest
Published: Jul 23, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source
const (
	DefaultMaxBufferedBytes            int64 = 1 << 20
	DefaultPruneLimit                        = 500
	MaxPruneLimit                            = 5000
	DefaultMaxTerminalRecordsPerPlugin       = 1000
	MaxTerminalRecordsPerPlugin              = 100_000
	DefaultTerminalRetention                 = 7 * 24 * time.Hour
	SessionRevokedReason                     = "session revoked"
)

Variables

View Source
var (
	ErrNotFound        = errors.New("plugin stream not found")
	ErrInvalidStream   = errors.New("plugin stream is invalid")
	ErrAlreadyExists   = errors.New("plugin stream already exists")
	ErrStreamClosed    = errors.New("plugin stream is closed")
	ErrStoreClosed     = errors.New("plugin stream store is closed")
	ErrStreamInvariant = errors.New("plugin stream storage invariant violated")
	ErrBackpressure    = errors.New("plugin stream backpressure limit exceeded")
	ErrDeliveryInvalid = errors.New("plugin stream delivery is invalid")
)

Functions

This section is empty.

Types

type AcknowledgeRequest added in v0.5.0

type AcknowledgeRequest struct {
	StreamID   string `json:"stream_id"`
	DeliveryID string `json:"delivery_id"`
}

type AppendRequest

type AppendRequest struct {
	StreamID string    `json:"stream_id"`
	Kind     string    `json:"kind,omitempty"`
	Data     []byte    `json:"data,omitempty"`
	Error    string    `json:"error,omitempty"`
	Now      time.Time `json:"-"`
}

type CloseRequest

type CloseRequest struct {
	StreamID    string                          `json:"stream_id"`
	Status      Status                          `json:"status,omitempty"`
	FailureCode capability.ExecutionFailureCode `json:"failure_code,omitempty"`
	Reason      string                          `json:"reason,omitempty"`
	Now         time.Time                       `json:"-"`
}

type DeliverRequest added in v0.5.0

type DeliverRequest struct {
	StreamID  string `json:"stream_id"`
	ReadID    string `json:"read_id"`
	MaxEvents int    `json:"max_events,omitempty"`
	MaxBytes  int64  `json:"max_bytes,omitempty"`
}

type Delivery added in v0.5.0

type Delivery struct {
	DeliveryID      string  `json:"delivery_id,omitempty"`
	ReadID          string  `json:"read_id"`
	StreamID        string  `json:"stream_id"`
	ThroughSequence uint64  `json:"through_sequence,omitempty"`
	Events          []Event `json:"events,omitempty"`
	Done            bool    `json:"done"`
	TerminalStatus  Status  `json:"terminal_status,omitempty"`
}

type Direction

type Direction string
const (
	DirectionRead   Direction = "read"
	DirectionWrite  Direction = "write"
	DirectionDuplex Direction = "duplex"
)

type Event

type Event struct {
	StreamID string    `json:"stream_id"`
	Sequence uint64    `json:"sequence"`
	Kind     string    `json:"kind"`
	Data     []byte    `json:"data,omitempty"`
	Error    string    `json:"error,omitempty"`
	At       time.Time `json:"at"`
}

type ListRequest added in v0.3.0

type ListRequest struct {
	PluginInstanceID string     `json:"plugin_instance_id,omitempty"`
	Owner            OwnerScope `json:"-"`
	AllOwners        bool       `json:"-"`
}

type MemoryStore

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

func NewMemoryStore

func NewMemoryStore() *MemoryStore

func (*MemoryStore) Acknowledge added in v0.5.0

func (s *MemoryStore) Acknowledge(_ context.Context, req AcknowledgeRequest) (Record, error)

func (*MemoryStore) Append

func (s *MemoryStore) Append(_ context.Context, req AppendRequest) (Event, error)

func (*MemoryStore) Close

func (s *MemoryStore) Close(_ context.Context, req CloseRequest) (Record, error)

func (*MemoryStore) Deliver added in v0.5.0

func (*MemoryStore) Durable added in v0.6.0

func (*MemoryStore) Durable() bool

func (*MemoryStore) Get

func (s *MemoryStore) Get(_ context.Context, streamID string) (Record, error)

func (*MemoryStore) List added in v0.3.0

func (s *MemoryStore) List(_ context.Context, req ListRequest) ([]Record, error)

func (*MemoryStore) MarkPluginTransition

func (*MemoryStore) Prune added in v0.5.0

func (*MemoryStore) Register

func (s *MemoryStore) Register(_ context.Context, req RegisterRequest) (Record, error)

func (*MemoryStore) RevokeSessionScope added in v0.6.0

func (*MemoryStore) Wait added in v0.5.0

func (s *MemoryStore) Wait(ctx context.Context, streamID string) error

type OwnerScope added in v0.5.0

type OwnerScope = capability.ExecutionOwnerScope

type PluginTransitionRequest

type PluginTransitionRequest struct {
	PluginInstanceID string                          `json:"plugin_instance_id"`
	ResourceScope    sessionctx.ResourceScope        `json:"-"`
	Status           Status                          `json:"status"`
	FailureCode      capability.ExecutionFailureCode `json:"failure_code,omitempty"`
	Reason           string                          `json:"reason,omitempty"`
	Now              time.Time                       `json:"-"`
}

type PluginTransitionResult added in v0.5.0

type PluginTransitionResult struct {
	Changed int `json:"changed"`
}

type PruneRequest added in v0.5.0

type PruneRequest struct {
	Before                      time.Time `json:"before"`
	Limit                       int       `json:"limit,omitempty"`
	MaxTerminalRecordsPerPlugin int       `json:"max_terminal_records_per_plugin,omitempty"`
}

type PruneResult added in v0.5.0

type PruneResult struct {
	Deleted int `json:"deleted"`
}

type Record

type Record struct {
	StreamID string `json:"stream_id"`
	capability.ExecutionBinding
	Direction        Direction                       `json:"direction"`
	Status           Status                          `json:"status"`
	FailureCode      capability.ExecutionFailureCode `json:"failure_code,omitempty"`
	Reason           string                          `json:"reason,omitempty"`
	ContentType      string                          `json:"content_type,omitempty"`
	MaxBufferedBytes int64                           `json:"max_buffered_bytes,omitempty"`
	BufferedBytes    int64                           `json:"buffered_bytes,omitempty"`
	CreatedAt        time.Time                       `json:"created_at"`
	UpdatedAt        time.Time                       `json:"updated_at"`
	ClosedAt         *time.Time                      `json:"closed_at,omitempty"`
}

type RegisterRequest

type RegisterRequest struct {
	StreamID         string                      `json:"stream_id"`
	ExecutionBinding capability.ExecutionBinding `json:"execution_binding"`
	Direction        Direction                   `json:"direction,omitempty"`
	ContentType      string                      `json:"content_type,omitempty"`
	MaxBufferedBytes int64                       `json:"max_buffered_bytes,omitempty"`
	Now              time.Time                   `json:"-"`
}

type RevokeSessionScopeRequest added in v0.6.0

type RevokeSessionScopeRequest struct {
	SessionScope sessionctx.SessionScope `json:"-"`
	Now          time.Time               `json:"-"`
}

type RevokeSessionScopeResult added in v0.6.0

type RevokeSessionScopeResult struct {
	Revoked int `json:"revoked"`
}

type SQLiteStore

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

func NewSQLiteStore

func NewSQLiteStore(ctx context.Context, path string) (*SQLiteStore, error)

func (*SQLiteStore) Acknowledge added in v0.5.0

func (s *SQLiteStore) Acknowledge(ctx context.Context, req AcknowledgeRequest) (Record, error)

func (*SQLiteStore) Append

func (s *SQLiteStore) Append(ctx context.Context, req AppendRequest) (Event, error)

func (*SQLiteStore) Close

func (s *SQLiteStore) Close(ctx context.Context, req CloseRequest) (Record, error)

func (*SQLiteStore) CloseDatabase

func (s *SQLiteStore) CloseDatabase() error

func (*SQLiteStore) Deliver added in v0.5.0

func (s *SQLiteStore) Deliver(ctx context.Context, req DeliverRequest) (Record, Delivery, error)

func (*SQLiteStore) Durable added in v0.6.0

func (*SQLiteStore) Durable() bool

func (*SQLiteStore) Get

func (s *SQLiteStore) Get(ctx context.Context, streamID string) (Record, error)

func (*SQLiteStore) List added in v0.3.0

func (s *SQLiteStore) List(ctx context.Context, req ListRequest) ([]Record, error)

func (*SQLiteStore) MarkPluginTransition

func (s *SQLiteStore) MarkPluginTransition(ctx context.Context, req PluginTransitionRequest) (PluginTransitionResult, error)

func (*SQLiteStore) Prune added in v0.5.0

func (s *SQLiteStore) Prune(ctx context.Context, req PruneRequest) (PruneResult, error)

func (*SQLiteStore) Register

func (s *SQLiteStore) Register(ctx context.Context, req RegisterRequest) (Record, error)

func (*SQLiteStore) RevokeSessionScope added in v0.6.0

func (*SQLiteStore) Wait added in v0.5.0

func (s *SQLiteStore) Wait(ctx context.Context, streamID string) error

type Status

type Status string
const (
	StatusOpen             Status = "open"
	StatusClosed           Status = "closed"
	StatusCanceled         Status = "canceled"
	StatusFailed           Status = "failed"
	StatusOrphanedDisabled Status = "orphaned_after_disable"
	StatusOrphanedRemoved  Status = "orphaned_after_uninstall"
)

type Store

type Store interface {
	Durable() bool
	Register(ctx context.Context, req RegisterRequest) (Record, error)
	List(ctx context.Context, req ListRequest) ([]Record, error)
	Get(ctx context.Context, streamID string) (Record, error)
	Append(ctx context.Context, req AppendRequest) (Event, error)
	Wait(ctx context.Context, streamID string) error
	Deliver(ctx context.Context, req DeliverRequest) (Record, Delivery, error)
	Acknowledge(ctx context.Context, req AcknowledgeRequest) (Record, error)
	Close(ctx context.Context, req CloseRequest) (Record, error)
	MarkPluginTransition(ctx context.Context, req PluginTransitionRequest) (PluginTransitionResult, error)
	RevokeSessionScope(ctx context.Context, req RevokeSessionScopeRequest) (RevokeSessionScopeResult, error)
	Prune(ctx context.Context, req PruneRequest) (PruneResult, error)
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL