lib/taskinator/persistence.rb in taskinator-0.3.3 vs lib/taskinator/persistence.rb in taskinator-0.3.5

@@ -1,18 +1,18 @@
 module Taskinator
   module Persistence
 
     class << self
+      def processes_list_key(scope=:shared)
+        "taskinator:#{scope}:processes"
+      end
+
       def add_process_to_list(process)
         Taskinator.redis do |conn|
-          conn.sadd "taskinator:#{list_key}", process.uuid
+          conn.sadd processes_list_key(process.scope), process.uuid
         end
       end
-
-      def list_key
-        'processes'
-      end
     end
 
     # mixin logic
     def self.included(klass)
       klass.class_eval do
@@ -25,11 +25,11 @@
 
       # class must override this method
       # to provide the base key to use for storing
       # it's instances, and it must be unique!
       def base_key
-        raise NotImplementedError
+        @base_key ||= 'shared'
       end
 
       # returns the storage key for the given identifier
       def key_for(uuid)
         "taskinator:#{base_key}:#{uuid}"
@@ -51,11 +51,11 @@
 
     module InstanceMethods
 
       def save
         Taskinator.redis do |conn|
-          conn.multi do
+          conn.pipelined do
             visitor = RedisSerializationVisitor.new(conn, self).visit
             conn.hmset(
               Taskinator::Process.key_for(uuid),
               :tasks_count,      visitor.task_count,
               :tasks_failed,     0,
@@ -192,10 +192,29 @@
           end
           yaml ? Taskinator::Persistence.deserialize(yaml) : {}
         end
       end
 
+      def cleanup
+        Taskinator.redis do |conn|
+
+          process_key = self.process_key
+
+          # delete processes/tasks data
+          conn.scan_each(:match => "#{process_key}:*", :count => 1000) do |key|
+            conn.del(key)
+          end
+
+          # remove the process
+          conn.del process_key
+
+          # remove from the list
+          conn.srem Persistence.processes_list_key(scope), uuid
+
+        end
+      end
+
     end
 
     class RedisSerializationVisitor < Visitor::Base
 
       #
@@ -281,9 +300,15 @@
       end
 
       def visit_args(attribute)
         values = @instance.send(attribute)
         yaml = Taskinator::Persistence.serialize(values)
+
+        # greater than 2 MB?
+        if (yaml.bytesize / (1024.0**2)) > 2
+          Taskinator.logger.warn("Large argument data detected for '#{self.to_s}'. Consider using intrinsic types instead, or try to reduce the amount of data provided.")
+        end
+
         @hmset += [attribute, yaml]
       end
 
       def task_count
         @task_count