registry

type Closeable ¶

type Closeable interface {
	Close() error
}

Closeable is implemented by nodes that hold resources (open connections, timers, file handles, ...) which must be released when their flow is undeployed. The engine calls Close exactly once per node instance during Undeploy, after the flow's context has been cancelled and any Start goroutine (see EmittingNode) has returned.

type ContextStore ¶

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

ContextStore is a thread-safe key-value store, mirroring Node-RED's flow/global context (`flow.get`/`flow.set`, `global.get`/`global.set`). The engine owns one global ContextStore shared by every flow; each active flow owns its own, private to that flow. Nodes reach both through NodeRuntime (see runtime.go).

func NewContextStore() *ContextStore

func (s *ContextStore) Delete(key string)

func (s *ContextStore) Get(key string) (interface{}, bool)

func (s *ContextStore) Keys() []string

func (s *ContextStore) Set(key string, value interface{})

type EagerlyReadyNode ¶

type EagerlyReadyNode interface {
	EmittingNode

	// EagerlyReady is a marker method with no meaningful behavior of its
	// own - implementing it opts a node into the engine's ready-wait (see
	// startEmittingNodes). The node's real obligation is to call
	// SignalReady(ctx) synchronously, as the first thing Start does,
	// once its setup is complete.
	EagerlyReady()
}

EagerlyReadyNode is an optional marker an EmittingNode additionally implements to tell the engine: "my Start does synchronous setup - e.g. subscribing to this flow's EventBus - that other nodes' very first Execute call might race against, so wait for it before returning from Deploy." Without this, Deploy returns as soon as Start's goroutine has merely been launched, not run; a message injected immediately afterward (the common "deploy, then inject" sequence in tests and in any caller that doesn't add its own delay) can reach a node that publishes an event - e.g. NodeRuntime.ReportStatus, or the engine's own automatic error/complete events after every Execute - before a Catch/ Status/Complete node's Start has actually called OnError/OnStatus/ OnComplete to subscribe, silently dropping it (registry.EventBus has no replay/buffering for a subscriber that arrives late).

Implementing this interface only makes sense for a node whose readiness depends on order relative to *other nodes in the same flow*, not on an external event source (a TCP listener, an MQTT subscription, a filesystem watch): those originate their own events independently of this flow's message processing, so there's nothing for them to race against in the same "Deploy then inject" sequence, and requiring the engine to wait for them would only add needless latency to every Deploy. Implemented by catch/status/complete; SignalReady/WithReady live in ready.go.

type EmittingNode ¶

type EmittingNode interface {
	NodeExecutor

	Start(ctx context.Context, emit func(payload map[string]interface{})) error
}

EmittingNode is implemented by nodes that originate messages on their own instead of only reacting to an incoming one (e.g. a TCP listener, an MQTT subscription, a file watcher). The engine invokes Start once, in its own goroutine, when the node's flow is deployed. Start must block until ctx is cancelled (which happens when the flow is undeployed) and call emit for every message it wants to inject at this node's output.

type EventBus ¶

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

EventBus is a simple, thread-safe pub/sub for flow-wide error, status, and completion events. Each active flow owns one. Handlers are called synchronously, in registration order, on the goroutine that published the event; handlers must not block for long or call back into the engine in a way that could deadlock (e.g. undeploying the same flow).

func NewEventBus() *EventBus

func (b *EventBus) OnComplete(handler func(NodeCompleteEvent))

func (b *EventBus) OnError(handler func(NodeErrorEvent))

func (b *EventBus) OnStatus(handler func(NodeStatusEvent))

func (b *EventBus) PublishComplete(evt NodeCompleteEvent)

func (b *EventBus) PublishError(evt NodeErrorEvent)

func (b *EventBus) PublishStatus(evt NodeStatusEvent)

type MultiOutputExecutor ¶

type MultiOutputExecutor interface {
	NodeExecutor

	// ExecuteMulti behaves like Execute but returns one payload per output
	// port ID (matching registry.Port.ID from the node's NodeMetadata.Outputs).
	// A port that is absent from the result, or mapped to nil, means: do not
	// send a message on that port for this invocation.
	ExecuteMulti(ctx interface{}, input map[string]interface{}) (map[string]map[string]interface{}, error)
}

MultiOutputExecutor is implemented by nodes that route different payloads to different output ports of the same node (e.g. a future Switch node picking one of several outputs, or a Catch/RBE node that may emit on no port at all for a given input). The engine calls ExecuteMulti instead of Execute when a node implements this interface.

type Node ¶

type Node struct {
	Type     string
	Metadata NodeMetadata
	Factory  NodeFactory
}

Node represents a registered node type.

type NodeCompleteEvent ¶

type NodeCompleteEvent struct {
	FlowID    string
	NodeID    string
	NodeType  string
	Payload   map[string]interface{}
	Timestamp time.Time
}

NodeCompleteEvent describes a node finishing a message successfully, published on a flow's EventBus so that Complete-style nodes can react to it. Payload is the message payload the node produced.

type NodeErrorEvent ¶

type NodeErrorEvent struct {
	FlowID    string
	NodeID    string
	NodeType  string
	Err       error
	Payload   map[string]interface{}
	Timestamp time.Time
}

NodeErrorEvent describes a node execution failure, published on a flow's EventBus so that Catch-style nodes (docs/NODE_PALETTE_PLAN.md, Phase 1) can react to errors from other nodes without a normal wire connection. Payload is the message payload the failing node was processing (not the full engine message - this package cannot depend on internal/engine without an import cycle, since engine already depends on registry for NodeExecutor and friends).

type NodeExecutor ¶

type NodeExecutor interface {
	// Execute processes the input message and returns output.
	// The context can be used for timeout and cancellation.
	Execute(ctx interface{}, input map[string]interface{}) (map[string]interface{}, error)

	// Validate checks the node configuration.
	// Should return an error if the configuration is invalid.
	Validate() error

	// GetConfig returns the current configuration as a map.
	GetConfig() map[string]interface{}

	// SetConfig sets the configuration from a map.
	// Should validate the configuration and return an error if invalid.
	SetConfig(config map[string]interface{}) error
}

NodeExecutor is the base interface for all nodes. Every node in Go—RED must implement this interface.

type NodeFactory ¶

type NodeFactory func() NodeExecutor

NodeFactory is a function that creates a new NodeExecutor instance.

type NodeMetadata ¶

type NodeMetadata struct {
	ID           string   `json:"id"`
	Type         string   `json:"type"`
	Name         string   `json:"name"`
	Description  string   `json:"description"`
	Category     string   `json:"category"`     // input, output, function, flow-control, storage
	Inputs       []Port   `json:"inputs"`       // Input ports
	Outputs      []Port   `json:"outputs"`      // Output ports
	ConfigSchema Schema   `json:"configSchema"` // Configuration schema
	Icon         string   `json:"icon"`         // SVG icon for UI
	Tags         []string `json:"tags"`         // Search tags
}

NodeMetadata contains metadata about a node type. This is used by the UI to display node information.

type NodeRegistry ¶

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

NodeRegistry manages all registered node types.

func GetGlobalRegistry() *NodeRegistry

func NewNodeRegistry() *NodeRegistry

func (r *NodeRegistry) GetAllNodes() []NodeMetadata

func (r *NodeRegistry) GetExecutor(nodeType string) (NodeExecutor, error)

func (r *NodeRegistry) GetMetadata(nodeType string) (NodeMetadata, error)

func (r *NodeRegistry) GetNodesByCategory(category string) []NodeMetadata

func (r *NodeRegistry) InitializeNode(nodeType string, config map[string]interface{}) (NodeExecutor, error)

func (r *NodeRegistry) IsRegistered(nodeType string) bool

func (r *NodeRegistry) RegisterFactory(nodeType string, factory NodeFactory, metadata NodeMetadata) error

func (r *NodeRegistry) RegisterNode(node *Node) error

func (r *NodeRegistry) Unregister(nodeType string) error

type NodeRuntime ¶

type NodeRuntime struct {
	FlowID        string
	NodeID        string
	NodeType      string
	FlowContext   *ContextStore
	GlobalContext *ContextStore
	// contains filtered or unexported fields
}

NodeRuntime gives a running node access to engine-level services that don't fit the input->output NodeExecutor.Execute contract: flow/global key-value context, error/status/complete event reporting and subscription (for Catch/Status/Complete nodes, docs/NODE_PALETTE_PLAN.md Phase 1), and submitting a message directly to another node in the same flow (for Link nodes). A node obtains it via RuntimeFromContext(ctx) inside Execute, ExecuteMulti, or Start - the engine embeds it into the context.Context it passes to all three.

func NewNodeRuntime(flowID, nodeID, nodeType string, flowContext, globalContext *ContextStore, events *EventBus, submit func(nodeID string, payload map[string]interface{}), getNode func(nodeID string) (NodeExecutor, bool)) *NodeRuntime

func RuntimeFromContext(ctx context.Context) (*NodeRuntime, bool)

func (r *NodeRuntime) GetNode(nodeID string) (NodeExecutor, bool)

func (r *NodeRuntime) OnComplete(handler func(NodeCompleteEvent))

func (r *NodeRuntime) OnError(handler func(NodeErrorEvent))

func (r *NodeRuntime) OnStatus(handler func(NodeStatusEvent))

func (r *NodeRuntime) ReportError(err error)

func (r *NodeRuntime) ReportStatus(status, detail string)

func (r *NodeRuntime) SubmitToNode(nodeID string, payload map[string]interface{})

type NodeStatusEvent ¶

type NodeStatusEvent struct {
	FlowID    string
	NodeID    string
	NodeType  string
	Status    string
	Detail    string
	Timestamp time.Time
}

NodeStatusEvent describes a node-reported status change (e.g. "connected", "disconnected", a progress message), published on a flow's EventBus so that Status-style nodes can react to it. Status is free-form and defined by the reporting node; the engine does not interpret it.

type Port ¶

type Port struct {
	ID          string `json:"id"`
	Name        string `json:"name"`
	Description string `json:"description"`
	Required    bool   `json:"required"`
}

Port defines an input or output port for a node.

type Property ¶

type Property struct {
	Type        string      `json:"type"` // string, number, boolean, object, array
	Description string      `json:"description"`
	Default     interface{} `json:"default"`
	Enum        []string    `json:"enum"` // Possible values for dropdowns
	Min         *float64    `json:"min"`
	Max         *float64    `json:"max"`
	Pattern     string      `json:"pattern"` // Regex pattern for strings
}

Property defines a configuration property for a node.

type Schema ¶

type Schema struct {
	Properties map[string]Property `json:"properties"`
	Required   []string            `json:"required"`
}

Schema defines the configuration schema for a node.