atp

Documentation

Overview

Package atp provides a client and server implementation of the Arcaflow Transport Protocol, which transports schema information and calls plugins.

Index

• Constants
• type Client
  • func NewClient(channel ClientChannel) Client
  • func NewClientWithLogger(channel ClientChannel, logger log.Logger) Client
• type ClientChannel
• type DecodedRuntimeMessage
• type ErrorMessage
  • func (e ErrorMessage) ToString(runID string) string
• type ExecutionResult
  • func NewErrorExecutionResult(err error) ExecutionResult
• type HelloMessage
• type RuntimeMessage
• type ServerError
  • func RunATPServer(ctx context.Context, stdin io.ReadCloser, stdout io.WriteCloser, ...) []*ServerError
  • func (e ServerError) String() string
• type SignalMessage
  • func (s SignalMessage) ToInput(runID string) schema.Input
• type WorkDoneMessage
• type WorkStartMessage

Constants

const (
	MessageTypeWorkStart  uint32 = 1
	MessageTypeWorkDone   uint32 = 2
	MessageTypeSignal     uint32 = 3
	MessageTypeClientDone uint32 = 4
	MessageTypeError      uint32 = 5
)

All messages that can be contained in a RuntimeMessage struct.

const ProtocolVersion int64 = 3

Types

type Client

type Client interface {
	// ReadSchema reads the schema from the ATP server.
	ReadSchema() (*schema.SchemaSchema, error)
	// Execute executes a step with a given context and returns the resulting output. Assumes you called ReadSchema first.
	Execute(input schema.Input, receivedSignals chan schema.Input, emittedSignals chan<- schema.Input) ExecutionResult
	Close() error
	Encoder() *cbor.Encoder
	Decoder() *cbor.Decoder
}

Client is the way to read information from the ATP server and then send a task to it in the form of a step. A step can only be sent once, but signals can be sent until the step is over. It is a single session.

func NewClient

func NewClient(
	channel ClientChannel,
) Client

NewClient creates a new ATP client (part of the engine code).

func NewClientWithLogger

func NewClientWithLogger(
	channel ClientChannel,
	logger log.Logger,
) Client

NewClientWithLogger creates a new ATP client (part of the engine code) with a logger.

type ClientChannel

type ClientChannel interface {
	io.Reader
	io.Writer
	io.Closer
}

ClientChannel holds the methods to talking to an ATP server (plugin).

type DecodedRuntimeMessage

type DecodedRuntimeMessage struct {
	MessageID      uint32          `cbor:"id"`
	RunID          string          `cbor:"run_id"`
	RawMessageData cbor.RawMessage `cbor:"data"`
}

type ErrorMessage

type ErrorMessage struct {
	Error       string `cbor:"error"`
	StepFatal   bool   `cbor:"step_fatal"`
	ServerFatal bool   `cbor:"server_fatal"`
}

func (ErrorMessage) ToString

func (e ErrorMessage) ToString(runID string) string

type ExecutionResult

type ExecutionResult struct {
	OutputID   string
	OutputData any
	Error      error
}

func NewErrorExecutionResult

func NewErrorExecutionResult(err error) ExecutionResult

type HelloMessage

type HelloMessage struct {
	Version int64 `cbor:"version"`
	Schema  any   `cbor:"schema"`
}

type RuntimeMessage

type RuntimeMessage struct {
	MessageID   uint32 `cbor:"id"`
	RunID       string `cbor:"run_id"`
	MessageData any    `cbor:"data"`
}

type ServerError

type ServerError struct {
	RunID       string
	Err         error
	StepFatal   bool
	ServerFatal bool
}

func RunATPServer

func RunATPServer(
	ctx context.Context,
	stdin io.ReadCloser,
	stdout io.WriteCloser,
	pluginSchema *schema.CallableSchema,
) []*ServerError

RunATPServer runs an ArcaflowTransportProtocol server with a given schema.

func (ServerError) String

func (e ServerError) String() string

type SignalMessage

type SignalMessage struct {
	SignalID string `cbor:"signal_id"`
	Data     any    `cbor:"data"`
}

func (SignalMessage) ToInput

func (s SignalMessage) ToInput(runID string) schema.Input

type WorkDoneMessage

type WorkDoneMessage struct {
	StepID     string `cbor:"step_id"`
	OutputID   string `cbor:"output_id"`
	OutputData any    `cbor:"output_data"`
	DebugLogs  string `cbor:"debug_logs"`
}

type WorkStartMessage

type WorkStartMessage struct {
	StepID string `cbor:"id"`
	Config any    `cbor:"config"`
}