Documentation
¶
Overview ¶
Package external provides the external plugin runtime for revoco.
Package external provides the external plugin runtime for revoco.
External plugins communicate via JSON-RPC 2.0 over stdin/stdout, allowing plugins to be written in any language.
Index ¶
- Constants
- func ExternalToDataItem(item *DataItem) *core.DataItem
- func FindPluginManifest(pluginDir string) (string, error)
- func RunPluginSetup(ctx context.Context, pluginDir string, setupCommands [][]string) error
- type BinaryDep
- type Client
- func (c *Client) Call(ctx context.Context, method string, params any) (any, error)
- func (c *Client) CallWithResult(ctx context.Context, method string, params any, result any) error
- func (c *Client) Close() error
- func (c *Client) Notifications() <-chan *Request
- func (c *Client) Notify(method string, params any) error
- type ConfigOption
- type DataItem
- type Error
- type ExternalConnector
- func (c *ExternalConnector) AuthType() string
- func (c *ExternalConnector) Capabilities() []core.ConnectorCapability
- func (c *ExternalConnector) Close() error
- func (c *ExternalConnector) ConfigSchema() []core.ConfigOption
- func (c *ExternalConnector) Delete(ctx context.Context, item core.DataItem) error
- func (c *ExternalConnector) Description() string
- func (c *ExternalConnector) FallbackOptions() []core.FallbackOption
- func (c *ExternalConnector) ID() string
- func (c *ExternalConnector) Initialize(ctx context.Context, cfg core.ConnectorConfig) error
- func (c *ExternalConnector) List(ctx context.Context, progress core.ProgressFunc) ([]core.DataItem, error)
- func (c *ExternalConnector) Name() string
- func (c *ExternalConnector) Plugin() *ExternalPlugin
- func (c *ExternalConnector) Read(ctx context.Context, item core.DataItem) (io.ReadCloser, error)
- func (c *ExternalConnector) ReadTo(ctx context.Context, item core.DataItem, destPath string, mode core.ImportMode) error
- func (c *ExternalConnector) RequiresAuth() bool
- func (c *ExternalConnector) SupportedDataTypes() []core.DataType
- func (c *ExternalConnector) TestConnection(ctx context.Context, cfg core.ConnectorConfig) error
- func (c *ExternalConnector) ValidateConfig(cfg core.ConnectorConfig) error
- func (c *ExternalConnector) Write(ctx context.Context, item core.DataItem, reader io.Reader) error
- func (c *ExternalConnector) WriteBatch(ctx context.Context, items []core.DataItem, ...) error
- func (c *ExternalConnector) WriteFrom(ctx context.Context, item core.DataItem, sourcePath string) error
- type ExternalConnectorPlugin
- type ExternalOutput
- func (o *ExternalOutput) CanOutput(item *core.DataItem) bool
- func (o *ExternalOutput) ConfigSchema() []svccore.ConfigOption
- func (o *ExternalOutput) Description() string
- func (o *ExternalOutput) Export(ctx context.Context, item svccore.ProcessedItem) error
- func (o *ExternalOutput) ExportBatch(ctx context.Context, items []svccore.ProcessedItem, ...) error
- func (o *ExternalOutput) Finalize(ctx context.Context) error
- func (o *ExternalOutput) ID() string
- func (o *ExternalOutput) Initialize(ctx context.Context, cfg svccore.OutputConfig) error
- func (o *ExternalOutput) Name() string
- func (o *ExternalOutput) Plugin() *ExternalPlugin
- func (o *ExternalOutput) Selector() *plugins.Selector
- func (o *ExternalOutput) SupportedItemTypes() []string
- type ExternalOutputPlugin
- type ExternalPlugin
- func (p *ExternalPlugin) Call(ctx context.Context, method string, params any) (any, error)
- func (p *ExternalPlugin) CallWithResult(ctx context.Context, method string, params any, result any) error
- func (p *ExternalPlugin) Client() *Client
- func (p *ExternalPlugin) Info() plugins.PluginInfo
- func (p *ExternalPlugin) IsLoaded() bool
- func (p *ExternalPlugin) Load(ctx context.Context) error
- func (p *ExternalPlugin) Reload(ctx context.Context) error
- func (p *ExternalPlugin) Selector() *plugins.Selector
- func (p *ExternalPlugin) Unload() error
- type ExternalProcessor
- func (p *ExternalProcessor) CanProcess(item *core.DataItem) bool
- func (p *ExternalProcessor) ConfigSchema() []plugins.ConfigOption
- func (p *ExternalProcessor) DefaultSelector() *plugins.Selector
- func (p *ExternalProcessor) Description() string
- func (p *ExternalProcessor) ID() string
- func (p *ExternalProcessor) Name() string
- func (p *ExternalProcessor) Plugin() *ExternalPlugin
- func (p *ExternalProcessor) Process(ctx context.Context, item *core.DataItem, config map[string]any) (*core.DataItem, error)
- func (p *ExternalProcessor) ProcessBatch(ctx context.Context, items []*core.DataItem, config map[string]any, ...) ([]*core.DataItem, error)
- type ExternalProcessorPlugin
- type InitializeParams
- type InitializeResult
- type PluginInfo
- type PluginManifest
- type Process
- func (p *Process) Client() *Client
- func (p *Process) ExitCode() int
- func (p *Process) ExitError() error
- func (p *Process) IsRunning() bool
- func (p *Process) PID() int
- func (p *Process) Start(ctx context.Context) error
- func (p *Process) Stderr() string
- func (p *Process) Stop(timeout time.Duration) error
- type ProcessManager
- func (pm *ProcessManager) Get(pluginID string) (*Process, bool)
- func (pm *ProcessManager) SetShutdownTimeout(d time.Duration)
- func (pm *ProcessManager) SetStartTimeout(d time.Duration)
- func (pm *ProcessManager) Start(ctx context.Context, pluginID string, command string, args []string, ...) (*Process, error)
- func (pm *ProcessManager) Stop(pluginID string) error
- func (pm *ProcessManager) StopAll() error
- type ProcessResult
- type Request
- type Response
- type SetupResult
Constants ¶
const ( ErrCodeParse = -32700 // Invalid JSON ErrCodeInvalidRequest = -32600 // Invalid request object ErrCodeMethodNotFound = -32601 // Method not found ErrCodeInvalidParams = -32602 // Invalid method parameters ErrCodeInternal = -32603 // Internal error // Custom error codes (application-defined, must be -32000 to -32099) ErrCodePluginError = -32000 // Plugin-specific error ErrCodeTimeout = -32001 // Operation timed out ErrCodeNotInitialized = -32002 // Plugin not initialized ErrCodeAuthRequired = -32003 // Authentication required )
Standard JSON-RPC 2.0 error codes
const ( // Lifecycle methods MethodInitialize = "initialize" MethodShutdown = "shutdown" // Info methods MethodGetInfo = "getInfo" MethodGetCapabilities = "getCapabilities" MethodGetConfigSchema = "getConfigSchema" // Connector methods MethodList = "list" MethodRead = "read" MethodWrite = "write" MethodDelete = "delete" MethodTestConnection = "testConnection" // Processor methods MethodCanProcess = "canProcess" MethodProcess = "process" MethodProcessBatch = "processBatch" // Output methods MethodCanOutput = "canOutput" MethodExport = "export" MethodExportBatch = "exportBatch" MethodFinalize = "finalize" )
Variables ¶
This section is empty.
Functions ¶
func ExternalToDataItem ¶
ExternalToDataItem converts an external protocol DataItem to core.DataItem.
func FindPluginManifest ¶
FindPluginManifest looks for a plugin manifest in a directory.
Types ¶
type BinaryDep ¶
type BinaryDep struct {
Name string `json:"name" yaml:"name"`
Command string `json:"command" yaml:"command"`
VersionFlag string `json:"versionFlag" yaml:"versionFlag"`
MinVersion string `json:"minVersion" yaml:"minVersion"`
InstallCommands map[string]string `json:"installCommands" yaml:"installCommands"` // os -> command
}
BinaryDep describes a required binary dependency.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client handles JSON-RPC communication with an external process.
func (*Client) CallWithResult ¶
CallWithResult calls a method and unmarshals the result into the given target.
func (*Client) Notifications ¶
Notifications returns the channel for receiving notifications from the plugin.
type ConfigOption ¶
type ConfigOption struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Type string `json:"type"`
Default any `json:"default,omitempty"`
Options []string `json:"options,omitempty"`
Required bool `json:"required,omitempty"`
Sensitive bool `json:"sensitive,omitempty"`
}
ConfigOption describes a configuration option.
type DataItem ¶
type DataItem struct {
ID string `json:"id"`
Type string `json:"type"`
Path string `json:"path"`
RemoteID string `json:"remoteId,omitempty"`
SourceConnID string `json:"sourceConnId,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
Size int64 `json:"size,omitempty"`
Checksum string `json:"checksum,omitempty"`
}
DataItem represents an item being processed.
func DataItemToExternal ¶
DataItemToExternal converts a core.DataItem to the external protocol format.
type Error ¶
type Error struct {
Code int `json:"code"`
Message string `json:"message"`
Data any `json:"data,omitempty"`
}
Error represents a JSON-RPC 2.0 error.
type ExternalConnector ¶
type ExternalConnector struct {
// contains filtered or unexported fields
}
ExternalConnector adapts an ExternalPlugin to the connector interfaces.
func NewExternalConnector ¶
func NewExternalConnector(plugin *ExternalPlugin) *ExternalConnector
NewExternalConnector creates a connector adapter for an external plugin.
func (*ExternalConnector) AuthType ¶
func (c *ExternalConnector) AuthType() string
func (*ExternalConnector) Capabilities ¶
func (c *ExternalConnector) Capabilities() []core.ConnectorCapability
func (*ExternalConnector) Close ¶
func (c *ExternalConnector) Close() error
func (*ExternalConnector) ConfigSchema ¶
func (c *ExternalConnector) ConfigSchema() []core.ConfigOption
func (*ExternalConnector) Description ¶
func (c *ExternalConnector) Description() string
func (*ExternalConnector) FallbackOptions ¶
func (c *ExternalConnector) FallbackOptions() []core.FallbackOption
func (*ExternalConnector) ID ¶
func (c *ExternalConnector) ID() string
func (*ExternalConnector) Initialize ¶
func (c *ExternalConnector) Initialize(ctx context.Context, cfg core.ConnectorConfig) error
func (*ExternalConnector) List ¶
func (c *ExternalConnector) List(ctx context.Context, progress core.ProgressFunc) ([]core.DataItem, error)
func (*ExternalConnector) Name ¶
func (c *ExternalConnector) Name() string
func (*ExternalConnector) Plugin ¶
func (c *ExternalConnector) Plugin() *ExternalPlugin
Plugin returns the underlying plugin.
func (*ExternalConnector) Read ¶
func (c *ExternalConnector) Read(ctx context.Context, item core.DataItem) (io.ReadCloser, error)
func (*ExternalConnector) ReadTo ¶
func (c *ExternalConnector) ReadTo(ctx context.Context, item core.DataItem, destPath string, mode core.ImportMode) error
func (*ExternalConnector) RequiresAuth ¶
func (c *ExternalConnector) RequiresAuth() bool
func (*ExternalConnector) SupportedDataTypes ¶
func (c *ExternalConnector) SupportedDataTypes() []core.DataType
func (*ExternalConnector) TestConnection ¶
func (c *ExternalConnector) TestConnection(ctx context.Context, cfg core.ConnectorConfig) error
func (*ExternalConnector) ValidateConfig ¶
func (c *ExternalConnector) ValidateConfig(cfg core.ConnectorConfig) error
func (*ExternalConnector) WriteBatch ¶
type ExternalConnectorPlugin ¶
type ExternalConnectorPlugin struct {
*ExternalPlugin
// contains filtered or unexported fields
}
ExternalConnectorPlugin wraps an ExternalPlugin as a ConnectorPlugin.
func NewExternalConnectorPlugin ¶
func NewExternalConnectorPlugin(plugin *ExternalPlugin) *ExternalConnectorPlugin
NewExternalConnectorPlugin creates a ConnectorPlugin from an ExternalPlugin.
func (*ExternalConnectorPlugin) AsConnector ¶
func (p *ExternalConnectorPlugin) AsConnector() core.Connector
func (*ExternalConnectorPlugin) AsReader ¶
func (p *ExternalConnectorPlugin) AsReader() (core.ConnectorReader, bool)
func (*ExternalConnectorPlugin) AsTester ¶
func (p *ExternalConnectorPlugin) AsTester() (core.ConnectorTester, bool)
func (*ExternalConnectorPlugin) AsWriter ¶
func (p *ExternalConnectorPlugin) AsWriter() (core.ConnectorWriter, bool)
type ExternalOutput ¶
type ExternalOutput struct {
// contains filtered or unexported fields
}
ExternalOutput adapts an ExternalPlugin to the Output interface.
func NewExternalOutput ¶
func NewExternalOutput(plugin *ExternalPlugin) *ExternalOutput
NewExternalOutput creates an output adapter for an external plugin.
func (*ExternalOutput) ConfigSchema ¶
func (o *ExternalOutput) ConfigSchema() []svccore.ConfigOption
func (*ExternalOutput) Description ¶
func (o *ExternalOutput) Description() string
func (*ExternalOutput) Export ¶
func (o *ExternalOutput) Export(ctx context.Context, item svccore.ProcessedItem) error
Export sends a single item to the destination.
func (*ExternalOutput) ExportBatch ¶
func (o *ExternalOutput) ExportBatch(ctx context.Context, items []svccore.ProcessedItem, progress svccore.ProgressFunc) error
ExportBatch sends multiple items to the destination.
func (*ExternalOutput) Finalize ¶
func (o *ExternalOutput) Finalize(ctx context.Context) error
Finalize completes the export process.
func (*ExternalOutput) ID ¶
func (o *ExternalOutput) ID() string
func (*ExternalOutput) Initialize ¶
func (o *ExternalOutput) Initialize(ctx context.Context, cfg svccore.OutputConfig) error
Initialize prepares the output for use.
func (*ExternalOutput) Name ¶
func (o *ExternalOutput) Name() string
func (*ExternalOutput) Plugin ¶
func (o *ExternalOutput) Plugin() *ExternalPlugin
Plugin returns the underlying plugin.
func (*ExternalOutput) Selector ¶
func (o *ExternalOutput) Selector() *plugins.Selector
func (*ExternalOutput) SupportedItemTypes ¶
func (o *ExternalOutput) SupportedItemTypes() []string
SupportedItemTypes returns the item types this output can handle.
type ExternalOutputPlugin ¶
type ExternalOutputPlugin struct {
*ExternalPlugin
// contains filtered or unexported fields
}
ExternalOutputPlugin wraps an ExternalPlugin as an OutputPlugin.
func NewExternalOutputPlugin ¶
func NewExternalOutputPlugin(plugin *ExternalPlugin) *ExternalOutputPlugin
NewExternalOutputPlugin creates an OutputPlugin from an ExternalPlugin.
func (*ExternalOutputPlugin) AsOutput ¶
func (p *ExternalOutputPlugin) AsOutput() svccore.Output
func (*ExternalOutputPlugin) Selector ¶
func (p *ExternalOutputPlugin) Selector() *plugins.Selector
type ExternalPlugin ¶
type ExternalPlugin struct {
// contains filtered or unexported fields
}
ExternalPlugin represents a loaded external plugin.
func LoadExternalPlugin ¶
func LoadExternalPlugin(manager *ProcessManager, dir string) (*ExternalPlugin, error)
LoadExternalPlugin loads an external plugin from a directory.
func (*ExternalPlugin) CallWithResult ¶
func (p *ExternalPlugin) CallWithResult(ctx context.Context, method string, params any, result any) error
CallWithResult is a helper method to call a method and unmarshal the result.
func (*ExternalPlugin) Client ¶
func (p *ExternalPlugin) Client() *Client
Client returns the JSON-RPC client for this plugin.
func (*ExternalPlugin) Info ¶
func (p *ExternalPlugin) Info() plugins.PluginInfo
Info returns the plugin metadata.
func (*ExternalPlugin) IsLoaded ¶
func (p *ExternalPlugin) IsLoaded() bool
IsLoaded returns true if the plugin process is running.
func (*ExternalPlugin) Load ¶
func (p *ExternalPlugin) Load(ctx context.Context) error
Load initializes the plugin (starts the process).
func (*ExternalPlugin) Reload ¶
func (p *ExternalPlugin) Reload(ctx context.Context) error
Reload reloads the plugin.
func (*ExternalPlugin) Selector ¶
func (p *ExternalPlugin) Selector() *plugins.Selector
Selector returns the default selector for this plugin.
func (*ExternalPlugin) Unload ¶
func (p *ExternalPlugin) Unload() error
Unload releases resources held by the plugin (stops the process).
type ExternalProcessor ¶
type ExternalProcessor struct {
// contains filtered or unexported fields
}
ExternalProcessor adapts an ExternalPlugin to the Processor interface.
func NewExternalProcessor ¶
func NewExternalProcessor(plugin *ExternalPlugin) *ExternalProcessor
NewExternalProcessor creates a processor adapter for an external plugin.
func (*ExternalProcessor) CanProcess ¶
func (p *ExternalProcessor) CanProcess(item *core.DataItem) bool
func (*ExternalProcessor) ConfigSchema ¶
func (p *ExternalProcessor) ConfigSchema() []plugins.ConfigOption
func (*ExternalProcessor) DefaultSelector ¶
func (p *ExternalProcessor) DefaultSelector() *plugins.Selector
func (*ExternalProcessor) Description ¶
func (p *ExternalProcessor) Description() string
func (*ExternalProcessor) ID ¶
func (p *ExternalProcessor) ID() string
func (*ExternalProcessor) Name ¶
func (p *ExternalProcessor) Name() string
func (*ExternalProcessor) Plugin ¶
func (p *ExternalProcessor) Plugin() *ExternalPlugin
Plugin returns the underlying plugin.
func (*ExternalProcessor) ProcessBatch ¶
type ExternalProcessorPlugin ¶
type ExternalProcessorPlugin struct {
*ExternalPlugin
// contains filtered or unexported fields
}
ExternalProcessorPlugin wraps an ExternalPlugin as a ProcessorPlugin.
func NewExternalProcessorPlugin ¶
func NewExternalProcessorPlugin(plugin *ExternalPlugin) *ExternalProcessorPlugin
NewExternalProcessorPlugin creates a ProcessorPlugin from an ExternalPlugin.
func (*ExternalProcessorPlugin) AsProcessor ¶
func (p *ExternalProcessorPlugin) AsProcessor() plugins.Processor
func (*ExternalProcessorPlugin) Selector ¶
func (p *ExternalProcessorPlugin) Selector() *plugins.Selector
type InitializeParams ¶
type InitializeParams struct {
Config map[string]any `json:"config"`
WorkDir string `json:"workDir"`
CacheDir string `json:"cacheDir"`
}
InitializeParams are the parameters for the initialize method.
type InitializeResult ¶
type InitializeResult struct {
Success bool `json:"success"`
Message string `json:"message,omitempty"`
}
InitializeResult is the result of the initialize method.
type PluginInfo ¶
type PluginInfo struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description"`
Version string `json:"version"`
Type string `json:"type"` // connector, processor, output
Capabilities []string `json:"capabilities,omitempty"`
DataTypes []string `json:"dataTypes,omitempty"`
RequiresAuth bool `json:"requiresAuth,omitempty"`
AuthType string `json:"authType,omitempty"`
}
PluginInfo is returned by getInfo.
type PluginManifest ¶
type PluginManifest struct {
// Basic info
ID string `json:"id" yaml:"id"`
Name string `json:"name" yaml:"name"`
Description string `json:"description" yaml:"description"`
Version string `json:"version" yaml:"version"`
Type string `json:"type" yaml:"type"` // connector, processor, output
// Execution
Command string `json:"command" yaml:"command"` // The command to run
Args []string `json:"args" yaml:"args"` // Arguments to the command
// Setup
SetupCommands [][]string `json:"setupCommands" yaml:"setupCommands"` // Commands to run on first setup
// Dependencies
BinaryDeps []BinaryDep `json:"binaryDeps" yaml:"binaryDeps"` // Required binaries
// Configuration
ConfigOptions []ConfigOption `json:"configOptions" yaml:"configOptions"`
// Capabilities (for connectors)
Capabilities []string `json:"capabilities" yaml:"capabilities"`
DataTypes []string `json:"dataTypes" yaml:"dataTypes"`
// Auth (for connectors)
RequiresAuth bool `json:"requiresAuth" yaml:"requiresAuth"`
AuthType string `json:"authType" yaml:"authType"`
// Selector (for processors)
Selector string `json:"selector" yaml:"selector"`
}
PluginManifest describes an external plugin.
type Process ¶
type Process struct {
// contains filtered or unexported fields
}
Process represents a running external plugin process.
func NewProcess ¶
func NewProcess(pluginID string, command string, args []string, env map[string]string, workDir string) (*Process, error)
NewProcess creates a new process (but doesn't start it).
type ProcessManager ¶
type ProcessManager struct {
// contains filtered or unexported fields
}
ProcessManager manages external plugin processes.
func NewProcessManager ¶
func NewProcessManager() *ProcessManager
NewProcessManager creates a new process manager.
func (*ProcessManager) Get ¶
func (pm *ProcessManager) Get(pluginID string) (*Process, bool)
Get returns a running process by plugin ID.
func (*ProcessManager) SetShutdownTimeout ¶
func (pm *ProcessManager) SetShutdownTimeout(d time.Duration)
SetShutdownTimeout sets the timeout for shutting down processes.
func (*ProcessManager) SetStartTimeout ¶
func (pm *ProcessManager) SetStartTimeout(d time.Duration)
SetStartTimeout sets the timeout for starting processes.
func (*ProcessManager) Start ¶
func (pm *ProcessManager) Start(ctx context.Context, pluginID string, command string, args []string, env map[string]string, workDir string) (*Process, error)
Start launches a new plugin process.
func (*ProcessManager) Stop ¶
func (pm *ProcessManager) Stop(pluginID string) error
Stop terminates a plugin process.
func (*ProcessManager) StopAll ¶
func (pm *ProcessManager) StopAll() error
StopAll terminates all plugin processes.
type ProcessResult ¶
type ProcessResult struct {
Item *DataItem `json:"item,omitempty"`
Skipped bool `json:"skipped,omitempty"`
Error string `json:"error,omitempty"`
}
ProcessResult is returned by process methods.
type Request ¶
type Request struct {
JSONRPC string `json:"jsonrpc"`
ID *int `json:"id,omitempty"` // nil for notifications
Method string `json:"method"`
Params any `json:"params,omitempty"`
}
Request represents a JSON-RPC 2.0 request.
type Response ¶
type Response struct {
JSONRPC string `json:"jsonrpc"`
ID *int `json:"id,omitempty"`
Result any `json:"result,omitempty"`
Error *Error `json:"error,omitempty"`
}
Response represents a JSON-RPC 2.0 response.
type SetupResult ¶
SetupResult contains the result of running a setup command.
func RunSetupCommand ¶
func RunSetupCommand(ctx context.Context, workDir string, command string, args []string) (*SetupResult, error)
RunSetupCommand runs a setup command for a plugin (e.g., pip install, npm install).