Refactor code to make it easier to extract functionality at a later date

lib/sequent/core/command_service.rb

@@ -49,13 +49,7 @@ def process_commands
           begin
             transaction_provider.transactional do
               while(!command_queue.empty?) do
-                command = command_queue.pop
-                filters.each { |filter| filter.execute(command) }
-
-                raise CommandNotValid.new(command) unless command.valid?
-                parsed_command = command.parse_attrs_to_correct_types
-                command_handlers.select { |h| h.class.handles_message?(parsed_command) }.each { |h| h.handle_message parsed_command }
-                repository.commit(parsed_command)
+                process_command(command_queue.pop)
               end
             end
           ensure
@@ -65,6 +59,15 @@ def process_commands
         end
       end
 
+      def process_command(command)
+        filters.each { |filter| filter.execute(command) }
+
+        raise CommandNotValid.new(command) unless command.valid?
+        parsed_command = command.parse_attrs_to_correct_types
+        command_handlers.select { |h| h.class.handles_message?(parsed_command) }.each { |h| h.handle_message parsed_command }
+        repository.commit(parsed_command)
+      end
+
       def command_queue
         Thread.current[:command_service_commands] ||= Queue.new
       end

lib/sequent/core/event_publisher.rb

@@ -35,20 +35,22 @@ def events_queue
 
       def process_events
         Sequent::Util.skip_if_already_processing(:events_queue_lock) do
-          while(!events_queue.empty?) do
-            event = events_queue.pop
-            configuration.event_handlers.each do |handler|
-              begin
-                handler.handle_message event
-              rescue
-                events_queue.clear
-                raise PublishEventError.new(handler.class, event)
-              end
+          begin
+            while(!events_queue.empty?) do
+              process_event(events_queue.pop)
             end
+          ensure
+            events_queue.clear
           end
         end
       end
 
+      def process_event(event)
+        configuration.event_handlers.each { |handler| handler.handle_message event }
+      rescue
+        raise PublishEventError.new(handler.class, event)
+      end
+
       def configuration
         Sequent.configuration
       end