harness

type Control ¶

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

func (ctrl *Control) GetOrLookupPlan(dc DataContext, unmarshal func(pbd *fnpb.ProcessBundleDescriptor) any) (any, error)

func (ctrl *Control) RegisterMonitor(dc DataContext, monFn Monitor)

func (ctrl *Control) RegisterSplitter(dc DataContext, splitFn Splitter)

type DataChannel ¶

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

DataChannel manages a single gRPC stream over the Data API. Data from multiple bundles can be multiplexed over this stream. Data is pushed over the channel, so data for a reader may arrive before the reader connects. Thread-safe.

func (c *DataChannel) OpenElementChan(ctx context.Context, ptransformID string, instID instructionID, expectedTimerTransforms []string) (<-chan Elements, error)

func (c *DataChannel) OpenTimerWrite(ctx context.Context, ptransformID string, instID instructionID, family string) io.WriteCloser

func (c *DataChannel) OpenWrite(ctx context.Context, ptransformID string, instID instructionID) io.WriteCloser

type DataChannelManager ¶

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

DataChannelManager manages data channels over the Data API. A fixed number of channels are generally used, each managing multiple logical byte streams. Thread-safe.

func (m *DataChannelManager) Open(ctx context.Context, port Port) (*DataChannel, error)

type DataContext ¶

type DataContext struct {
	Data  *ScopedDataManager
	State *ScopedStateManager
	// contains filtered or unexported fields
}

DataContext holds connectors to various data connections, incl. state and side input.

type Elements ¶

type Elements struct {
	Data, Timers                []byte
	TimerFamilyID, PtransformID string
}

Elements holds data or timers sent across the data channel. If TimerFamilyID is populated, it's a timer, otherwise it's data elements.

type ExecFunc ¶

type ExecFunc func(context.Context, *Control, DataContext) (*fnpb.ProcessBundleResponse, error)

type Monitor ¶

type Monitor func() (map[string]*pipepb.MonitoringInfo, map[string][]byte)

Monitor is a function that returns any new labels, and the set of payloads being returned to the runner.

type NextBuffer ¶

type NextBuffer interface {
	NextBuf() ([]byte, error)
	Close() error
}

type Options ¶

type Options struct {
	RunnerCapabilities []string // URNs for what runners are able to understand over the FnAPI.
	LoggingEndpoint    string   // Endpoint for remote logging.
	StatusEndpoint     string   // Endpoint for worker status reporting.
}

Options for harness.Main that affect execution of the harness, such as runner capabilities.

type Port ¶

type Port struct {
	URL string
}

Port represents the connection port of external operations.

type ScopedDataManager ¶

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

ScopedDataManager scopes the global gRPC data manager to a single instruction. The indirection makes it easier to control access.

func NewScopedDataManager(mgr *DataChannelManager, instID instructionID) *ScopedDataManager

func (s *ScopedDataManager) Close() error

func (s *ScopedDataManager) OpenElementChan(ctx context.Context, id StreamID, expectedTimerTransforms []string) (<-chan Elements, error)

func (s *ScopedDataManager) OpenTimerWrite(ctx context.Context, id StreamID, family string) (io.WriteCloser, error)

func (s *ScopedDataManager) OpenWrite(ctx context.Context, id StreamID) (io.WriteCloser, error)

type ScopedStateManager ¶

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

ScopedStateManager scopes the global gRPC state manager to a single instruction for side input use. The indirection makes it easier to control access.

func NewScopedStateManager(mgr *StateChannelManager, instID instructionID) *ScopedStateManager

func (s *ScopedStateManager) Close() error

func (s *ScopedStateManager) OpenReader(ctx context.Context, url string, key *fnpb.StateKey) (NextBuffer, error)

func (s *ScopedStateManager) OpenWriter(ctx context.Context, url string, key *fnpb.StateKey, wt writeTypeEnum) (io.Writer, error)

type Splitter ¶

type Splitter func(map[string]*fnpb.ProcessBundleSplitRequest_DesiredSplit) (*fnpb.ProcessBundleSplitResponse, error)

Splitter is a function that requests a split from a given set of plans.

type StateChannel ¶

type StateChannel struct {
	DoneCh <-chan struct{}
	// contains filtered or unexported fields
}

StateChannel manages state transactions over a single gRPC connection. It does not need to track readers and writers as carefully as the DataChannel, because the state protocol is request-based.

func (c *StateChannel) Send(req *fnpb.StateRequest) (*fnpb.StateResponse, error)

type StateChannelManager ¶

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

StateChannelManager manages data channels over the State API. A fixed number of channels are generally used, each managing multiple logical byte streams. Thread-safe.

func (m *StateChannelManager) Open(ctx context.Context, url string) (*StateChannel, error)

type StreamID ¶

type StreamID struct {
	Port         Port
	PtransformID string
}

StreamID represents the static information needed to identify a data stream. Dynamic information, notably bundleID, is provided implicitly by the managers.