journal.rb in openwferu-0.9.11

- old
+ new

@@ -105,15 +105,16 @@
             # flushed.
             #
             def queue_event (event, *args)
                 synchronize do
 
+                    #ldebug { "queue_event() :#{event}" }
+
                     return if event == :stop
                     return if event == :launch
                     return if event == :reschedule
 
-                    #wfid = args[0].parent_wfid
                     wfid = extract_fei(args[0]).parent_wfid
                         #
                         # maybe args[0] could be a FlowExpression instead
                         # of a FlowExpressionId instance
                     #puts "___#{event}__wfid : #{wfid}"
@@ -121,25 +122,27 @@
                     e = serialize_event(event, *args)
 
                     bucket = get_bucket(wfid) 
                     bucket << e
 
+                    #ldebug { "queue_event() bucket : #{bucket.object_id}" }
+
                     if event == :terminate
 
                         bucket.flush
                         @buckets.delete(wfid)
 
                         if @application_context[:keep_journals] == true
                             #
                             # 'move' journal to the done/ subdir of journal/
                             #
                             FileUtils.cp(
-                                bucket.get_path,
-                                @donedir + "/" + bucket.get_file_name)
+                                bucket.path,
+                                @donedir + "/" + File.basename(bucket.path))
                         end
 
-                        FileUtils.rm bucket.get_path
+                        FileUtils.rm bucket.path
                     end
                 end
             end
 
             #
@@ -161,31 +164,34 @@
                 linfo { "flush_buckets() flushed #{count} buckets" } \
                     if count > 0
             end
 
             def get_bucket (wfid)
-                @buckets[wfid] ||= Bucket.new(@workdir, wfid)
+                @buckets[wfid] ||= Bucket.new(get_path(wfid))
             end
 
             def serialize_event (event, *args)
                 args.insert(0, event)
                 args.insert(1, Time.now)
                 args.to_yaml
             end
 
+            def get_path (wfid)
+                @workdir + "/" + wfid.to_s + ".journal"
+            end
+
             #
             # for each process instance, there is one bucket holding the
             # events waiting to get written in the journal
             #
             class Bucket
 
-                attr_reader :wfid, :events
+                attr_reader :path, :events
 
-                def initialize (workdir, wfid)
+                def initialize (path)
                     super()
-                    @workdir = workdir
-                    @wfid = wfid
+                    @path = path
                     @events = []
                 end
 
                 def << (event)
                     @events << event
@@ -195,23 +201,15 @@
                     @events.size
                 end
                 alias :length :size
 
                 def flush
-                    File.open(get_path, "w+") do |f|
+                    File.open(@path, "a+") do |f|
                         @events.each do |e|
                             f.puts e
                         end
                     end
                     @events.clear
-                end
-
-                def get_path
-                    @workdir + "/" + get_file_name
-                end
-
-                def get_file_name
-                    @wfid.to_s + ".journal"
                 end
             end
 
     end
 