connection_graph.rb

# frozen_string_literal: true

module Syskit
    # Represents the actual connection graph between task context proxies.
    # Its vertices are instances of Orocos::TaskContext, and edges are
    # mappings from [source_port_name, sink_port_name] pairs to the
    # connection policy between these ports.
    #
    # {Runtime::ActualDataFlow} is the actual global graph instance
    # in which the overall system connections are maintained in practice.
    # Its vertices are Orocos::TaskContext and the edge information a
    # mapping of the form
    #
    #   (source_port_name, sink_port_name) => policy
    #
    # {Runtime::RequiredDataFlow} is the graph instance that manages concrete
    # connections between Syskit::TaskContext (i.e. where all the forwarding
    # through compositions has been removed). It is updated as needed by
    # {Runtime::ConnectionManagement#update_required_dataflow_graph}. Its
    # vertices are Syskit::TaskContext instances and the edge information a
    # mapping of the form
    #
    #   (source_port_name, sink_port_name) => policy
    #
    # Syskit::Flows::DataFlow is the flow graph that is generated by the engine.
    # Its vertices are Syskit::Component, it contains "virtual" connections
    # (connections between composition ports and task ports). The edge
    # information is of the form
    #
    #   (source_port_name, sink_port_name) => policy
    #
    # Since compositions are still in this graph, source_port_name and
    # sink_port_name can be either outputs or inputs. In all the other
    # graphs, they are guaranteed to be output ports resp. input ports.
    class ConnectionGraph < Roby::Relations::Graph
        # Needed for Roby's marshalling (so that we can dump the connection
        # graph as a constant)
        attr_accessor :name

        # Tests if +port+, which has to be an output port, is connected
        def has_out_connections?(task, port)
            each_out_neighbour(task) do |child_task|
                if edge_info(task, child_task).each_key.any? { |source_port, _| source_port == port }
                    return true
                end
            end
            false
        end

        # Tests if +port+, which has to be an input port, is connected
        def has_in_connections?(task, port)
            each_in_neighbour(task) do |parent_task|
                if edge_info(parent_task, task).each_key.any? { |_, target_port| target_port == port }
                    return true
                end
            end
            false
        end

        # Tests if there is a connection between +source_task+:+source_port+
        # and +sink_task+:+sink_port+
        def connected?(source_task, source_port, sink_task, sink_port)
            unless has_edge?(source_task, sink_task)
                return false
            end

            edge_info(source_task, sink_task).key?([source_port, sink_port])
        end

        def add_connections(source_task, sink_task, mappings) # :nodoc:
            add_edge(source_task, sink_task, mappings)
        end

        def freeze_edge_info(mappings)
            mappings.each do |pair, policy|
                pair.freeze
                policy.freeze
            end.freeze
        end

        def add_edge(source_task, sink_task, mappings)
            if mappings.empty?
                raise ArgumentError, "the connection set is empty"
            end

            freeze_edge_info(mappings)
            super
        end

        def merge_info(source, sink, current_mappings, additional_mappings)
            merged = current_mappings.merge(additional_mappings) do |k, v1, v2|
                if v1 != v2
                    raise ArgumentError, "cannot override policy information by default: trying to override the policy between #{source} and #{sink} from #{k}: #{v1} to #{k}: #{v2}"
                end

                v1
            end
            freeze_edge_info(merged)
            merged
        end

        # Removes the given set of connections between +source_task+ and
        # +sink_task+.
        #
        # +mappings+ is an array of port name pairs [output_port_name,
        # input_port_name]
        def remove_connections(source_task, sink_task, mappings) # :nodoc:
            current_mappings = edge_info(source_task, sink_task).dup
            mappings.each do |source_port, sink_port|
                current_mappings.delete([source_port, sink_port])
            end
            if current_mappings.empty?
                remove_relation(source_task, sink_task)
                remove_vertex(source_task) if leaf?(source_task) && root?(source_task)
                remove_vertex(sink_task)   if leaf?(sink_task)   && root?(sink_task)
            else
                # To make the relation system call #update_info
                set_edge_info(source_task, sink_task, current_mappings)
            end
        end

        # Yield or enumerates incoming connections
        #
        # @param [Component] component the component whose connections we want
        #   to enumerate
        # @param [#name,String,nil] port if non-nil, the port for
        #   which we want to enumerate the connections (in which case
        #   the sink_port yield parameter is guaranteed to be this name).
        #   Otherwise, all ports are enumerated.
        #
        # @yield each connections
        # @yieldparam [Syskit::TaskContext] source_task the source task in
        #   the connection
        # @yieldparam [String] source_port the source port name on source_task
        # @yieldparam [String] sink_port the sink port name on self. If
        #   the port argument is non-nil, it is guaranteed to be the
        #   same.
        # @yieldparam [Hash] policy the connection policy
        #
        # @see each_concrete_input_connection each_concrete_output_connection
        #   each_output_connection
        def each_in_connection(task, port = nil)
            return enum_for(__method__, task, port) unless block_given?

            if port.respond_to? :name
                port = port.name
            end

            each_in_neighbour(task) do |source_task|
                edge_info(source_task, task).each do |(source_port, sink_port), policy|
                    if !port || (sink_port == port)
                        yield(source_task, source_port, sink_port, policy)
                    end
                end
            end
        end

        # Yield or enumerates the connections that exist from the output
        # ports of self.
        #
        # @param [Component] source_task the task whose connections are being
        #   enumerated
        # @param [#name,String,nil] port if non-nil, the port for
        #   which we want to enumerate the connections (in which case
        #   the source_port yield parameter is guaranteed to be this name).
        #   Otherwise, all ports are enumerated.
        #
        # @yield each connections
        # @yieldparam [String] source_port the source port name on self. If
        #   the port argument is non-nil, it is guaranteed to be the
        #   same.
        # @yieldparam [String] sink_port the sink port name on sink_task.
        # @yieldparam [Syskit::Component] sink_task the sink task in
        #   the connection
        # @yieldparam [Hash] policy the connection policy
        #
        def each_out_connection(task, port = nil)
            return enum_for(__method__, task, port) unless block_given?

            if port.respond_to? :name
                port = port.name
            end

            each_out_neighbour(task) do |sink_task|
                edge_info(task, sink_task).each do |(source_port, sink_port), policy|
                    if !port || (port == source_port)
                        yield(source_port, sink_port, sink_task, policy)
                    end
                end
            end
            self
        end
    end
end