<commit_before>module Pacer
  module Routes
    module RouteOperations
      def join(name, options = {}, &block)
        args = { :transform => :join,
          element_type: :vertex,
          graph: options.fetch(:multi_graph, Pacer::MultiGraph.new),
          from_graph: graph
        }
        args[:multi_graph] = options[:multi_graph] if options[:multi_graph]
        route = chain_route(args).route.join(options.fetch(:key, :key)) { |v| v }
        route.join(name, &block)
      end

      def collect_with(multigraph)
        vars[:multigraph] = multigraph
      end

      def collect_as(name, opts = {})
        process do |element|
          g = vars[:multigraph] ||= Pacer::MultiGraph.new
          v = vars[name] = g.create_vertex
          v[name] = element
          within = opts[:within]
          if within
            within_v = vars[within]
            g.create_edge nil, v, within_v, :within
          end
        end
      end

      def add_to(collection_name, name = nil)
        process do |element|
          v = vars[collection_name]
          if name
            existing = v[name]
            if existing
              existing << element
            else
              v[name] = [element]
            end
          end
          if block_given?
            yield v, element
          end
        end
      end

      def reduce_to(collection_name, name, starting_value)
        process do |element|
          v = vars[collection_name]
          total = v[name]
          if total
            v[name] = yield total, element
          else
            v[name] = yield starting_value, element
          end
        end
      end

      def execute!
        p = pipe
        while p.hasNext
          p.next<commit_msg>map_to makes sense, too.<commit_after>module Pacer
  module Routes
    module RouteOperations
      def join(name, options = {}, &block)
        args = { :transform => :join,
          element_type: :vertex,
          graph: options.fetch(:multi_graph, Pacer::MultiGraph.new),
          from_graph: graph
        }
        args[:multi_graph] = options[:multi_graph] if options[:multi_graph]
        route = chain_route(args).route.join(options.fetch(:key, :key)) { |v| v }
        route.join(name, &block)
      end

      def collect_with(multigraph)
        vars[:multigraph] = multigraph
      end

      def collect_as(name, opts = {})
        process do |element|
          g = vars[:multigraph] ||= Pacer::MultiGraph.new
          v = vars[name] = g.create_vertex
          v[name] = element
          within = opts[:within]
          if within
            within_v = vars[within]
            g.create_edge nil, v, within_v, :within
          end
        end
      end

      def add_to(collection_name, name = nil)
        process do |element|
          v = vars[collection_name]
          if name
            existing = v[name]
            if existing
              existing << element
            else
              v[name] = [element]
            end
          end
          if block_given?
            yield v, element
          end
        end
      end

      def map_to(collection_name, name)
        process do |element|
          v = vars[collection_name]
          existing = v[name]
          if block_given?
            mapped = yield element, v
          else
            mapped = element
          end
          if existing
            existing << mapped
          else
            v[name] = [mapped]
          end
        end
      end

      def reduce_to(collection_name, name, starting_value)
        process do |element|
          v = vars[collection_name]
          total = v[name]
          if total
            v[name] = yield total, element
          else
            v[name] = yield starting_value, element
          end
        end
      end

      def execute!
        p = pipe
        while p.hasNext
          p.next
{'fraction_non_alphanumeric': 0.08515003847140293, 'fraction_numerical': 0.0, 'mean_word_length': 1.4652338811630847, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '495941', 'n_tokens_mistral': 1085, 'n_tokens_neox': 1062, 'n_words': 286}