lib/rbbt/persist.rb in rbbt-util-5.13.31 vs lib/rbbt/persist.rb in rbbt-util-5.13.32

@@ -109,20 +109,12 @@
         res = Open.read(path).split("\n", -1)
         res.pop
         res
       when :marshal
         Open.open(path) do |stream|
-          case stream
-          when StringIO
-            begin
-              Marshal.load(stream) 
-            rescue
-              raise $!
-            end
-          else
-            Marshal.load(stream)
-          end
+          content = stream.read.unpack("m").first
+          Marshal.load(content) 
         end
       when :yaml
         Open.open(path) do |stream|
           YAML.load(stream)
         end
@@ -178,11 +170,12 @@
         Misc.sensiblewrite(path, content.to_s)
       end
     when :marshal_tsv
       Misc.sensiblewrite(path, Marshal.dump(content.dup))
     when :marshal
-      Misc.sensiblewrite(path, Marshal.dump(content))
+      dump = Marshal.dump(content)
+      Misc.sensiblewrite(path, [dump].pack("m"))
     when :yaml
       Misc.sensiblewrite(path, YAML.dump(content))
     when :float, :integer, :tsv
       Misc.sensiblewrite(path, content.to_s)
     else
@@ -286,103 +279,38 @@
     stream = res.stream if res.respond_to? :stream
 
     if stream
       if persist_options[:no_load] == :stream 
         res = tee_stream(stream, path, type, stream.respond_to?(:callback)? stream.callback : nil, stream.respond_to?(:abort_callback)? stream.abort_callback : nil)
+        res.lockfile = lockfile
 
-        ConcurrentStream.setup res do
-          begin
-            lockfile.unlock #if File.exists? lockfile.path and lockfile.locked?
-          rescue Exception
-            Log.medium "Lockfile exception: " << $!.message
-          end
-        end
-        res.abort_callback = Proc.new do
-          begin
-            lockfile.unlock #if File.exists? lockfile.path and lockfile.locked?
-          rescue Exception
-            Log.medium "Lockfile exception: " << $!.message
-          end
-        end
         raise KeepLocked.new res 
       else
+        stream = res.get_stream if res.respond_to? :get_stream
         begin
-          res = case type
-                when :array
-                  res.read.split "\n"
-                when :tsv
-                  TSV.open(res)
-                else
-                  res.read
-                end
-          res.join if res.respond_to? :join
-          res
+          Open.write(path, stream)
+          Open.open(path) do |stream|
+            case type
+            when :array
+              stream.read.split "\n"
+            when :tsv
+              TSV.open(stream)
+            else
+              stream.read
+            end
+          end
         rescue
-          res.abort if res.respond_to? :abort
+          stream.abort if stream.respond_to? :abort
           raise $!
         end
       end
     else
       res
     end
   end
 
-  def self._get_result(path, type, persist_options, lockfile, &block)
-    res = yield
-
-    if persist_options[:no_load] == :stream 
-      stream = IO === res ? res : res.stream
-      res = tee_stream(stream, path, type, stream.respond_to?(:callback)? stream.callback : nil, stream.respond_to?(:abort_callback)? stream.abort_callback : nil)
-
-      ConcurrentStream.setup res do
-        begin
-            lockfile.unlock #if File.exists? lockfile.path and lockfile.locked?
-        rescue Exception
-          Log.medium "Lockfile exception: " << $!.message
-        end
-      end
-      res.abort_callback = Proc.new do
-        begin
-          lockfile.unlock #if File.exists? lockfile.path and lockfile.locked?
-        rescue Exception
-          Log.medium "Lockfile exception: " << $!.message
-        end
-      end
-      raise KeepLocked.new res 
-    end
-
-    case res
-    when IO
-      begin
-        res = case
-              when :array
-                res.read.split "\n"
-              when :tsv
-                TSV.open(res)
-              else
-                res.read
-              end
-        res.join if res.respond_to? :join
-      rescue
-        res.abort if res.respond_to? :abort
-        raise $!
-      end
-    when (defined? TSV and TSV::Dumper)
-      begin
-        io = res.stream
-        res = TSV.open(io)
-        io.join if io.respond_to? :join
-      rescue
-        io.abort if io.respond_to? :abort
-        raise $!
-      end
-    end
-    res
-  end
-
   def self.persist_file(path, type, persist_options, &block)
-
     begin
       if is_persisted?(path, persist_options)
         Log.low "Persist up-to-date: #{ path } - #{Misc.fingerprint persist_options}"
         return path if persist_options[:no_load]
         return load_file(path, type) 
@@ -412,10 +340,12 @@
 
         Misc.lock(path) do
           save_file(path, type, res)
         end
 
-        persist_options[:no_load] ? path : res
+        return path if persist_options[:no_load]
+
+        res
       end
 
     rescue Lockfile::StolenLockError
       Log.medium "Lockfile stolen: #{path}"
       retry