MCPX MCP Gateway

package streamflow import ( internaltypes "lunar/engine/streams/internal-types" ) // Ensure interfaces are implemented var _ internaltypes.ConnectionEdgeI = &ConnectionEdge{} type ExternalEdge struct { stream internaltypes.StreamRefI // For stream connections flow internaltypes.FlowRefI // For flow connections } func (ee ExternalEdge) equal(other ExternalEdge) bool { // Case 1: Both are stream connections if ee.stream != nil && other.stream != nil { return ee.stream.GetName() == other.stream.GetName() && ee.stream.GetAt() == other.stream.GetAt() } // Case 2: Both are flow connections if ee.flow != nil && other.flow != nil { return ee.flow.GetName() == other.flow.GetName() && ee.flow.GetAt() == other.flow.GetAt() } return false } // ConnectionEdge represents an edge in the DAG. // It defines the connection between two processors or between a processor and a stream/flow. type ConnectionEdge struct { ExternalEdge node *FlowGraphNode // Target node for processor-to-processor connections condition string // Condition under which this edge is followed, optional } // NewConnectionEdge creates a new connection edge. func NewConnectionEdge(condition string) *ConnectionEdge { return &ConnectionEdge{condition: condition} } // GetCondition returns the condition for the connection edge. func (ce *ConnectionEdge) GetCondition() string { return ce.condition } // GetTargetNode returns the target node for the connection edge. func (ce *ConnectionEdge) GetTargetNode() internaltypes.FlowGraphNodeI { return ce.node } // GetTargetStream returns the target stream for the connection edge. func (ce *ConnectionEdge) GetTargetStream() internaltypes.StreamRefI { return ce.stream } // GetTargetFlow returns the target flow for the connection edge. func (ce *ConnectionEdge) GetTargetFlow() internaltypes.FlowRefI { return ce.flow } // IsNodeAvailable checks if the target node is available for the connection edge. func (ce *ConnectionEdge) IsNodeAvailable() bool { return ce.node != nil } // IsStreamAvailable checks if the target stream is available for the connection edge. func (ce *ConnectionEdge) IsStreamAvailable() bool { return ce.stream != nil } // IsFlowAvailable checks if the target flow is available for the connection edge. func (ce *ConnectionEdge) IsFlowAvailable() bool { return ce.flow != nil } // IsValid checks if any target is defined for the connection edge. func (ce *ConnectionEdge) IsValid() bool { return ce.node != nil || ce.stream != nil || ce.flow != nil } func (ce *ConnectionEdge) equal(other *ConnectionEdge) bool { if ce.condition != other.condition { return false } // compare external edges if ce.ExternalEdge.equal(other.ExternalEdge) { return true } // compare target nodes if ce.node != nil && other.node != nil && ce.node.equal(other.node) { return true } return false }