lib/mongoid/contextual/map_reduce.rb in mongoid-5.4.1 vs lib/mongoid/contextual/map_reduce.rb in mongoid-6.0.0.beta

@@ -30,17 +30,16 @@
       #
       # @return [ Enumerator ] The enumerator.
       #
       # @since 3.0.0
       def each
-        validate_out!
         if block_given?
-          @map_reduce.each do |doc|
+          documents.each do |doc|
             yield doc
           end
         else
-          @map_reduce.to_enum
+          to_enum
         end
       end
 
       # Get the number of documents emitted by the map/reduce.
       #
@@ -63,11 +62,11 @@
       #
       # @return [ MapReduce ] The map reduce.
       #
       # @since 3.0.0
       def finalize(function)
-        @map_reduce = @map_reduce.finalize(function)
+        command[:finalize] = function
         self
       end
 
       # Initialize the new map/reduce directive.
       #
@@ -78,13 +77,14 @@
       # @param [ String ] map The map js function.
       # @param [ String ] reduce The reduce js function.
       #
       # @since 3.0.0
       def initialize(collection, criteria, map, reduce)
-        @collection = collection
-        @criteria = criteria
-        @map_reduce = @criteria.view.map_reduce(map, reduce)
+        @collection, @criteria = collection, criteria
+        command[:mapreduce] = collection.name.to_s
+        command[:map], command[:reduce] = map, reduce
+        apply_criteria_options
       end
 
       # Get the number of documents that were input into the map/reduce.
       #
       # @example Get the count of input documents.
@@ -104,16 +104,15 @@
       #
       # @return [ MapReduce ] The map/reduce.
       #
       # @since 3.0.0
       def js_mode
-        @map_reduce = @map_reduce.js_mode(true)
+        command[:jsMode] = true
         self
       end
 
       # Specifies where the map/reduce output is to be stored.
-      # Please see MongoDB documentation for supported map reduce options.
       #
       # @example Store output in memory.
       #   map_reduce.out(inline: 1)
       #
       # @example Store output in a collection, replacing existing documents.
@@ -123,24 +122,21 @@
       #   map_reduce.out(merge: "collection_name")
       #
       # @example Store output in a collection, reducing existing documents.
       #   map_reduce.out(reduce: "collection_name")
       #
-      # @example Return results from map reduce.
-      #   map_reduce.out(inline: 1)
-      #
       # @param [ Hash ] location The place to store the results.
       #
       # @return [ MapReduce ] The map/reduce object.
       #
       # @since 3.0.0
       def out(location)
         normalized = location.dup
         normalized.update_values do |value|
           value.is_a?(::Symbol) ? value.to_s : value
         end
-        @map_reduce = @map_reduce.out(normalized)
+        command[:out] = normalized
         self
       end
 
       # Get the number of documents output by the map/reduce.
       #
@@ -161,16 +157,12 @@
       #
       # @return [ Hash ] The raw output.
       #
       # @since 3.0.0
       def raw
-        validate_out!
-        cmd = command
-        opts = { read: cmd.delete(:read).options } if cmd[:read]
-        @map_reduce.database.command(cmd, opts || {}).first
+        results
       end
-      alias :results :raw
 
       # Execute the map/reduce, returning the raw output.
       # Useful when you don't care about map/reduce's ouptut.
       #
       # @example Run the map reduce
@@ -202,11 +194,11 @@
       #
       # @return [ MapReduce ]
       #
       # @since 3.0.0
       def scope(object)
-        @map_reduce = @map_reduce.scope(object)
+        command[:scope] = object
         self
       end
 
       # Get the execution time of the map/reduce.
       #
@@ -238,17 +230,98 @@
   finalize: #{command[:finalize]}
   out:      #{command[:out].inspect}>
 }
       end
 
-      def command
-        @map_reduce.send(:map_reduce_spec)[:selector]
+      private
+
+      # Apply criteria specific options - query, sort, limit.
+      #
+      # @api private
+      #
+      # @example Apply the criteria options
+      #   map_reduce.apply_criteria_options
+      #
+      # @return [ nil ] Nothing.
+      #
+      # @since 3.0.0
+      def apply_criteria_options
+        command[:query] = criteria.selector
+        if sort = criteria.options[:sort]
+          command[:sort] = sort
+        end
+        if limit = criteria.options[:limit]
+          command[:limit] = limit
+        end
       end
 
-      private
+      # Get the result documents from the map/reduce. If the output was inline
+      # then we grab them from the results key. If the output was a temp
+      # collection then we need to execute a find on that collection.
+      #
+      # @api private
+      #
+      # @example Get the documents.
+      #   map_reduce.documents
+      #
+      # @return [ Array, Cursor ] The documents.
+      #
+      # @since 3.0.0
+      def documents
+        return results["results"] if results.has_key?("results")
+        view = client[output_collection].find
+        view.no_cursor_timeout if criteria.options[:timeout] == false
+        view
+      end
 
-      def validate_out!
-        raise Errors::NoMapReduceOutput.new({}) unless @map_reduce.out
+      # Get the collection that the map/reduce results were stored in.
+      #
+      # @api private
+      #
+      # @example Get the output collection.
+      #   map_reduce.output_collection
+      #
+      # @return [ Symbol, String ] The output collection.
+      #
+      # @since 3.0.0
+      def output_collection
+        command[:out].values.first
+      end
+
+      # Execute the map/reduce command and get the results.
+      #
+      # @api private
+      #
+      # @example Get the results.
+      #   map_reduce.results
+      #
+      # @return [ Hash ] The results of the command.
+      #
+      # @since 3.0.0
+      def results
+        raise Errors::NoMapReduceOutput.new(command) unless command[:out]
+        @results ||= __client__.command(command).first
+      end
+
+      # Get the client with the proper consistency.
+      #
+      # @api private
+      #
+      # @note We can use eventual if the output is set to inline.
+      #
+      # @example Get the client.
+      #   map_reduce.__client__
+      #
+      # @return [ Mongo::Client ] The client with consistency set.
+      #
+      # @since 3.0.15
+      def __client__
+        if command[:out][:inline] != 1
+          # @todo: close
+          client.with(read: { mode: :primary })
+        else
+          client
+        end
       end
     end
   end
 end