lib/rflow/component/port.rb in rflow-1.0.0a4 vs lib/rflow/component/port.rb in rflow-1.0.0a5

@@ -81,9 +81,29 @@
       def add_connection(key, connection)
         RFlow.logger.debug "Attaching #{connection.class.name} connection '#{connection.name}' (#{connection.uuid}) to port '#{name}' (#{uuid}), key '#{connection.input_port_key}'"
         connections_for[key] << connection
       end
 
+      # Removes a connection from a given key
+      def remove_connection(key, connection)
+        RFlow.logger.debug "Removing #{connection.class.name} connection '#{connection.name}' (#{connection.uuid}) from port '#{name}' (#{uuid}), key '#{connection.input_port_key}'"
+        connections_for[key].delete(connection)
+      end
+
+      def collect_messages(key, receiver)
+        begin
+          connection = RFlow::MessageCollectingConnection.new.tap do |c|
+            c.messages = receiver
+          end
+          add_connection key, connection
+
+          yield if block_given?
+          connection
+        ensure
+          remove_connection key, connection if connection && block_given?
+        end
+      end
+
       def direct_connect(other_port)
         case other_port
         when InputPort; add_connection nil, ForwardToInputPort.new(other_port)
         when OutputPort; add_connection nil, ForwardToOutputPort.new(other_port)
         else raise ArgumentError, "Unknown port type #{other_port.class.name}"