RUBY-1279 Refactor Operations

Author: estolfo

lib/mongo/auth/user/view.rb

@@ -44,7 +44,7 @@ class View
         def create(user_or_name, options = {})
           user = generate(user_or_name, options)
           client.send(:with_session, options) do |session|
-            Operation::Write::Command::CreateUser.new(
+            Operation::CreateUser.new(
               user: user,
               db_name: database.name,
               session: session
@@ -79,7 +79,7 @@ def initialize(database)
         # @since 2.0.0
         def remove(name, options = {})
           client.send(:with_session, options) do |session|
-            Operation::Write::Command::RemoveUser.new(
+            Operation::RemoveUser.new(
               user_name: name,
               db_name: database.name,
               session: session
@@ -103,7 +103,7 @@ def remove(name, options = {})
         def update(user_or_name, options = {})
           client.send(:with_session, options) do |session|
             user = generate(user_or_name, options)
-            Operation::Write::Command::UpdateUser.new(
+            Operation::UpdateUser.new(
               user: user,
               db_name: database.name,
               session: session
@@ -132,7 +132,7 @@ def info(name, options = {})
 
         def user_query(name, options = {})
           client.send(:with_session, options) do |session|
-            Operation::Commands::UsersInfo.new(
+            Operation::UsersInfo.new(
               user_name: name,
               db_name: database.name,
               session: session

lib/mongo/bulk_write.rb

@@ -61,7 +61,7 @@ def execute
             write_with_retry(session, write_concern) do |server, txn_num|
               execute_operation(
                   operation.keys.first,
-                  operation.values.first,
+                  operation.values.flatten,
                   server,
                   operation_id,
                   result_combiner,
@@ -72,7 +72,7 @@ def execute
             legacy_write_with_retry do |server|
               execute_operation(
                   operation.keys.first,
-                  operation.values.first,
+                  operation.values.flatten,
                   server,
                   operation_id,
                   result_combiner,
@@ -175,7 +175,8 @@ def execute_operation(name, values, server, operation_id, combiner, session, txn
         if values.size > server.max_write_batch_size
           split_execute(name, values, server, operation_id, combiner, session, txn_num)
         else
-          combiner.combine!(send(name, values, server, operation_id, session, txn_num), values.size)
+          r = send(name, values, server, operation_id, session, txn_num)
+          combiner.combine!(r, values.size)
         end
       rescue Error::MaxBSONSize, Error::MaxMessageSize => e
         raise e if values.size <= 1
@@ -195,35 +196,29 @@ def split_execute(name, values, server, operation_id, combiner, session, txn_num
     end
 
     def delete_one(documents, server, operation_id, session, txn_num)
-      Operation::Write::Bulk::Delete.new(
-        base_spec(operation_id, session).merge(:deletes => documents, :txn_num => txn_num)
-      ).execute(server)
+      spec = base_spec(operation_id, session).merge(:deletes => documents, :txn_num => txn_num)
+      Operation::Delete.new(spec).bulk_execute(server)
     end
 
     def delete_many(documents, server, operation_id, session, txn_num)
-      Operation::Write::Bulk::Delete.new(
-          base_spec(operation_id, session).merge(:deletes => documents)
-      ).execute(server)
+      spec = base_spec(operation_id, session).merge(:deletes => documents)
+      Operation::Delete.new(spec).bulk_execute(server)
     end
 
-
     def insert_one(documents, server, operation_id, session, txn_num)
-      Operation::Write::Bulk::Insert.new(
-        base_spec(operation_id, session).merge(:documents => documents, :txn_num => txn_num)
-      ).execute(server)
+      spec = base_spec(operation_id, session).merge(:documents => documents, :txn_num => txn_num)
+      Operation::Insert.new(spec).bulk_execute(server)
     end
 
     def update_one(documents, server, operation_id, session, txn_num)
-      Operation::Write::Bulk::Update.new(
-        base_spec(operation_id, session).merge(:updates => documents, :txn_num => txn_num)
-      ).execute(server)
+      spec = base_spec(operation_id, session).merge(:updates => documents, :txn_num => txn_num)
+      Operation::Update.new(spec).bulk_execute(server)
     end
     alias :replace_one :update_one
 
     def update_many(documents, server, operation_id, session, txn_num)
-      Operation::Write::Bulk::Update.new(
-          base_spec(operation_id, session).merge(:updates => documents)
-      ).execute(server)
+      spec = base_spec(operation_id, session).merge(:updates => documents)
+      Operation::Update.new(spec).bulk_execute(server)
     end
   end
 end

lib/mongo/cluster/reapers/cursor_reaper.rb

@@ -124,7 +124,7 @@ def kill_cursors
             if server.features.find_command_enabled?
               Cursor::Builder::KillCursorsCommand.update_cursors(op_spec, active_cursors_copy.to_a)
               if Cursor::Builder::KillCursorsCommand.get_cursors_list(op_spec).size > 0
-                Operation::Commands::Command.new(op_spec).execute(server)
+                Operation::KillCursors.new(op_spec).execute(server)
               end
             else
               Cursor::Builder::OpKillCursors.update_cursors(op_spec, active_cursors_copy.to_a)

lib/mongo/collection.rb

@@ -189,12 +189,12 @@ def create(opts = {})
         raise Error::UnsupportedCollation.new
       end
       client.send(:with_session, opts) do |session|
-        Operation::Commands::Create.new({
-                                          selector: operation,
-                                          db_name: database.name,
-                                          write_concern: write_concern,
-                                          session: session
-                                        }).execute(server)
+        Operation::Create.new({
+                                selector: operation,
+                                db_name: database.name,
+                                write_concern: write_concern,
+                                session: session
+                                }).execute(server)
       end
     end
 
@@ -215,12 +215,12 @@ def create(opts = {})
     # @since 2.0.0
     def drop(opts = {})
       client.send(:with_session, opts) do |session|
-        Operation::Commands::Drop.new({
-                                        selector: { :drop => name },
-                                        db_name: database.name,
-                                        write_concern: write_concern,
-                                        session: session
-                                      }).execute(next_primary)
+        Operation::Drop.new({
+                              selector: { :drop => name },
+                              db_name: database.name,
+                              write_concern: write_concern,
+                              session: session
+                              }).execute(next_primary)
       end
     rescue Error::OperationFailure => ex
       raise ex unless ex.message =~ /ns not found/
@@ -420,7 +420,7 @@ def inspect
     def insert_one(document, opts = {})
       client.send(:with_session, opts) do |session|
         write_with_retry(session, write_concern) do |server, txn_num|
-          Operation::Write::Insert.new(
+          Operation::Insert.new(
               :documents => [ document ],
               :db_name => database.name,
               :coll_name => name,

lib/mongo/collection/view/aggregation.rb

@@ -103,7 +103,7 @@ def new(options)
         end
 
         def initial_query_op(session)
-          Operation::Commands::Aggregate.new(aggregate_spec(session))
+          Operation::Aggregate.new(aggregate_spec(session))
         end
 
         def valid_server?(server)

lib/mongo/collection/view/iterable.rb

@@ -65,15 +65,15 @@ def initial_query_op(server, session)
           if server.features.find_command_enabled?
             initial_command_op(session)
           else
-            Operation::Read::Query.new(Builder::OpQuery.new(self).specification)
+            Operation::Find.new(Builder::OpQuery.new(self).specification)
           end
         end
 
         def initial_command_op(session)
           if explained?
-            Operation::Commands::Explain.new(Builder::FindCommand.new(self, session).explain_specification)
+            Operation::Explain.new(Builder::FindCommand.new(self, session).explain_specification)
           else
-            Operation::Commands::Find.new(Builder::FindCommand.new(self, session).specification)
+            Operation::Find.new(Builder::FindCommand.new(self, session).specification)
           end
         end
 

lib/mongo/collection/view/map_reduce.rb

@@ -217,7 +217,7 @@ def new(options)
         end
 
         def initial_query_op(session)
-          Operation::Commands::MapReduce.new(map_reduce_spec(session))
+          Operation::MapReduce.new(map_reduce_spec(session))
         end
 
         def valid_server?(server)
@@ -247,9 +247,9 @@ def find_command_spec(session)
 
         def fetch_query_op(server, session)
           if server.features.find_command_enabled?
-            Operation::Commands::Find.new(find_command_spec(session))
+            Operation::Find.new(find_command_spec(session))
           else
-            Operation::Read::Query.new(fetch_query_spec)
+            Operation::Find.new(fetch_query_spec)
           end
         end
 

lib/mongo/collection/view/readable.rb

@@ -139,13 +139,13 @@ def count(opts = {})
             server = selector.select_server(cluster)
             apply_collation!(cmd, server, opts)
             with_session(opts) do |session|
-              Operation::Commands::Count.new({
-                                                   :selector => cmd,
-                                                   :db_name => database.name,
-                                                   :options => {:limit => -1},
-                                                   :read => read_pref,
-                                                   :session => session
-                                               }).execute(server)
+              Operation::Count.new({
+                                     :selector => cmd,
+                                     :db_name => database.name,
+                                     :options => {:limit => -1},
+                                     :read => read_pref,
+                                     :session => session
+                                    }).execute(server)
             end.n.to_i
           end
         end
@@ -179,13 +179,13 @@ def distinct(field_name, opts = {})
             server = selector.select_server(cluster)
             apply_collation!(cmd, server, opts)
             with_session(opts) do |session|
-              Operation::Commands::Distinct.new({
-                                                   :selector => cmd,
-                                                   :db_name => database.name,
-                                                   :options => {:limit => -1},
-                                                   :read => read_pref,
-                                                   :session => session
-                                               }).execute(server)
+              Operation::Distinct.new({
+                                        :selector => cmd,
+                                        :db_name => database.name,
+                                        :options => {:limit => -1},
+                                        :read => read_pref,
+                                        :session => session
+                                       }).execute(server)
             end.first['values']
           end
         end
@@ -480,7 +480,7 @@ def server_selector
         def parallel_scan(cursor_count, options = {})
           session = client.send(:get_session, @options)
           server = server_selector.select_server(cluster)
-          cmd = Operation::Commands::ParallelScan.new({
+          cmd = Operation::ParallelScan.new({
                   :coll_name => collection.name,
                   :db_name => database.name,
                   :cursor_count => cursor_count,
@@ -489,14 +489,14 @@ def parallel_scan(cursor_count, options = {})
                 }.merge!(options))
           cmd.execute(server).cursor_ids.map do |cursor_id|
             result = if server.features.find_command_enabled?
-                       Operation::Commands::GetMore.new({
+                       Operation::GetMore.new({
                          :selector => {:getMore => cursor_id,
                                        :collection => collection.name},
                          :db_name => database.name,
                          :session => session
                        }).execute(server)
                      else
-                       Operation::Read::GetMore.new({
+                       Operation::GetMore.new({
                          :to_return => 0,
                          :cursor_id => cursor_id,
                          :db_name => database.name,

lib/mongo/collection/view/writable.rb

@@ -49,7 +49,7 @@ def find_one_and_delete(opts = {})
           with_session(opts) do |session|
             write_with_retry(session, write_concern) do |server, txn_num|
               apply_collation!(cmd, server, opts)
-              Operation::Commands::Command.new(
+              Operation::Command.new(
                   :selector => cmd,
                   :db_name => database.name,
                   :session => session,
@@ -121,7 +121,7 @@ def find_one_and_update(document, opts = {})
             write_with_retry(session, write_concern) do |server, txn_num|
               apply_collation!(cmd, server, opts)
               apply_array_filters!(cmd, server, opts)
-              Operation::Commands::Command.new(
+              Operation::Command.new(
                   :selector => cmd,
                   :db_name => database.name,
                   :session => session,
@@ -149,8 +149,8 @@ def delete_many(opts = {})
           with_session(opts) do |session|
             legacy_write_with_retry do |server|
               apply_collation!(delete_doc, server, opts)
-              Operation::Write::Delete.new(
-                  :delete => delete_doc,
+              Operation::Delete.new(
+                  :deletes => [ delete_doc ],
                   :db_name => collection.database.name,
                   :coll_name => collection.name,
                   :write_concern => collection.write_concern,
@@ -178,8 +178,8 @@ def delete_one(opts = {})
           with_session(opts) do |session|
             write_with_retry(session, write_concern) do |server, txn_num|
               apply_collation!(delete_doc, server, opts)
-              Operation::Write::Delete.new(
-                  :delete => delete_doc,
+              Operation::Delete.new(
+                  :deletes => [ delete_doc ],
                   :db_name => collection.database.name,
                   :coll_name => collection.name,
                   :write_concern => write_concern,
@@ -217,8 +217,8 @@ def replace_one(replacement, opts = {})
               apply_collation!(update_doc, server, opts)
               apply_array_filters!(update_doc, server, opts)
 
-              Operation::Write::Update.new(
-                  :update => update_doc,
+              Operation::Update.new(
+                  :updates => [ update_doc ],
                   :db_name => collection.database.name,
                   :coll_name => collection.name,
                   :write_concern => write_concern,
@@ -256,8 +256,8 @@ def update_many(spec, opts = {})
             legacy_write_with_retry do |server|
               apply_collation!(update_doc, server, opts)
               apply_array_filters!(update_doc, server, opts)
-              Operation::Write::Update.new(
-                  :update => update_doc,
+              Operation::Update.new(
+                  :updates => [ update_doc ],
                   :db_name => collection.database.name,
                   :coll_name => collection.name,
                   :write_concern => collection.write_concern,
@@ -296,8 +296,8 @@ def update_one(spec, opts = {})
               apply_collation!(update_doc, server, opts)
               apply_array_filters!(update_doc, server, opts)
 
-              Operation::Write::Update.new(
-                  :update => update_doc,
+              Operation::Update.new(
+                  :updates => [ update_doc ],
                   :db_name => collection.database.name,
                   :coll_name => collection.name,
                   :write_concern => write_concern,

lib/mongo/cursor.rb

@@ -206,10 +206,11 @@ def get_more
 
     def get_more_operation
       if @server.features.find_command_enabled?
-        Operation::Commands::GetMore.new(Builder::GetMoreCommand.new(self, @session).specification)
+        spec = Builder::GetMoreCommand.new(self, @session).specification
       else
-        Operation::Read::GetMore.new(Builder::OpGetMore.new(self).specification)
+        spec = Builder::OpGetMore.new(self).specification
       end
+      Operation::GetMore.new(spec)
     end
 
     def kill_cursors
@@ -227,11 +228,7 @@ def end_session
     end
 
     def kill_cursors_operation
-      if @server.features.find_command_enabled?
-        Operation::Commands::Command.new(kill_cursors_op_spec)
-      else
-        Operation::KillCursors.new(kill_cursors_op_spec)
-      end
+      Operation::KillCursors.new(kill_cursors_op_spec)
     end
 
     def kill_cursors_op_spec