lib/mongo/database/view.rb in mongo-2.9.2 vs lib/mongo/database/view.rb in mongo-2.10.0.rc0

@@ -22,10 +22,12 @@
       extend Forwardable
       include Enumerable
       include Retryable
 
       def_delegators :@database, :cluster, :read_preference, :client
+      # @api private
+      def_delegators :@database, :server_selector, :read_concern
       def_delegators :cluster, :next_primary
 
       # @return [ Integer ] batch_size The size of the batch of results
       #   when sending the listCollections command.
       attr_reader :batch_size
@@ -36,12 +38,12 @@
       # @return [ Collection ] collection The command collection.
       attr_reader :collection
 
       # Get all the names of the non-system collections in the database.
       #
-      # @example Get the collection names.
-      #   database.collection_names
+      # @note The set of returned collection names depends on the version of
+      #   MongoDB server that fulfills the request.
       #
       # @param [ Hash ] options Options for the listCollections command.
       #
       # @option options [ Integer ] :batch_size  The batch size for results
       #   returned from the listCollections command.
@@ -50,34 +52,40 @@
       #
       # @since 2.0.0
       def collection_names(options = {})
         @batch_size = options[:batch_size]
         session = client.send(:get_session, options)
-        cursor = read_with_retry_cursor(session, ServerSelector.get(mode: :primary), self) do |server|
+        cursor = read_with_retry_cursor(session, ServerSelector.primary, self) do |server|
           send_initial_query(server, session, name_only: true)
         end
         cursor.map do |info|
           if cursor.server.features.list_collections_enabled?
-            info[Database::NAME]
+            info['name']
           else
-            (info[Database::NAME] &&
-              info[Database::NAME].sub("#{@database.name}.", ''))
+            (info['name'] &&
+              info['name'].sub("#{@database.name}.", ''))
           end
+        end.reject do |name|
+          name.start_with?('system.') || name.include?('$')
         end
       end
 
       # Get info on all the collections in the database.
       #
+      # @note The set of collections returned, and the schema of the
+      #   information hash per collection, depends on the MongoDB server
+      #   version that fulfills the request.
+      #
       # @example Get info on each collection.
       #   database.list_collections
       #
       # @return [ Array<Hash> ] Info for each collection in the database.
       #
       # @since 2.0.5
       def list_collections
         session = client.send(:get_session)
-        collections_info(session, ServerSelector.get(mode: :primary))
+        collections_info(session, ServerSelector.primary)
       end
 
       # Create the new database view.
       #
       # @example Create the new database view.
@@ -91,21 +99,56 @@
         @batch_size =  nil
         @limit = nil
         @collection = @database[Database::COMMAND]
       end
 
+      # @api private
+      attr_reader :database
+
+      # Execute an aggregation on the database view.
+      #
+      # @example Aggregate documents.
+      #   view.aggregate([
+      #     { "$listLocalSessions" => {} }
+      #   ])
+      #
+      # @param [ Array<Hash> ] pipeline The aggregation pipeline.
+      # @param [ Hash ] options The aggregation options.
+      #
+      # @return [ Aggregation ] The aggregation object.
+      #
+      # @since 2.10.0
+      # @api private
+      def aggregate(pipeline, options = {})
+        Collection::View::Aggregation.new(self, pipeline, options)
+      end
+
       private
 
       def collections_info(session, server_selector, options = {}, &block)
+        description = nil
         cursor = read_with_retry_cursor(session, server_selector, self) do |server|
+          # TODO take description from the connection used to send the query
+          # once https://jira.mongodb.org/browse/RUBY-1601 is fixed.
+          description = server.description
           send_initial_query(server, session, options)
         end
-        if block_given?
-          cursor.each do |doc|
-            yield doc
+        # On 3.0+ servers, we get just the collection names.
+        # On 2.6 server, we get collection names prefixed with the database
+        # name. We need to filter system collections out here because
+        # in the caller we don't know which server version executed the
+        # command and thus what the proper filtering logic should be
+        # (it is valid for collection names to have dots, thus filtering out
+        # collections named system.* here for 2.6 servers would actually
+        # filter out collections in the system database).
+        if description.server_version_gte?('3.0')
+          cursor.reject do |doc|
+            doc['name'].start_with?('system.') || doc['name'].include?('$')
           end
         else
-          cursor.to_enum
+          docs = cursor.reject do |doc|
+            doc['name'].start_with?("#{database.name}.system") || doc['name'].include?('$')
+          end
         end
       end
 
       def collections_info_spec(session, options = {})
         { selector: {