change version to 0.2.9; save-document as blocking command

project.clj

-(defproject wwsoftware/document-storage "0.2.8"
+(defproject wwsoftware/document-storage "0.2.9"
   :description "FIXME: write description"
   :repositories [["internal" {:url "https://leukipp.fh-muenster.de:9082/releases"
                               :username "anonymous"}]]

src/document_storage/postgres/documents.clj

@@ -2,8 +2,8 @@
   (:require [document-storage.postgres.encoding :as encoding]
             [document-storage.postgres.cache :as cache]
             [editscript.core :as escript]
-            [clj-helper.vector :as vhelper]))
-
+            [clj-helper.vector :as vhelper])
+  (:import [java.util.concurrent ArrayBlockingQueue]))
 
 (defn get-document-frame-by-version [storage document-id version repository]
   (.run-query storage :get-document-frame-by-version
@@ -89,15 +89,22 @@
                  :reference (:id last-frame) ;;;FIXME save ID in reference not version
                  :data diff-doc})))
 
-(defn save-document [storage document-id document repository options]
-  (cache/update-query-cache-document storage repository document-id document)
-  (if-let [last-version (get-last-document-version storage document-id repository)]
-    (if (or (true? (:force-keyframe options))
-            (= (:data-type last-version) :delete))
-      (save-keyframe storage document-id document (:version last-version) repository)
-      (save-pframe storage document-id document (:version last-version) repository))
-    (save-keyframe storage document-id document -1 repository)))
+(def get-repo-queue (memoize (fn [reponame]
+                               (new ArrayBlockingQueue 1 true))))
 
+(defn save-document [storage document-id document repository options]
+  (let [queue (get-repo-queue repository)
+        job-id (gensym)]
+    (.put queue job-id)
+    (cache/update-query-cache-document storage repository document-id document)
+    (let [save-result (if-let [last-version (get-last-document-version storage document-id repository)]
+                        (if (or (true? (:force-keyframe options))
+                                (= (:data-type last-version) :delete))
+                          (save-keyframe storage document-id document (:version last-version) repository)
+                          (save-pframe storage document-id document (:version last-version) repository))
+                        (save-keyframe storage document-id document -1 repository))]
+      (.take queue)
+      save-result)))
 
 (defn delete-document [storage document-id repository]
   (.run-query storage :create-document-frame!

test/document_storage/postgres/core_test.clj

@@ -6,7 +6,9 @@
             [fipp.edn :refer [pprint] :rename {pprint fipp}]
             [document-storage.protocol :as dsprot]
             [environ.core :refer [env]]
-            [document-storage.protocol-test :as protocol-tests]))
+            [document-storage.protocol-test :as protocol-tests]
+            [document-storage.testdata :refer [generate-document]]
+            [clj-helper.string :as shelper]))
 
 (def ds-atom (atom nil))
 
@@ -61,3 +63,39 @@
       (is (= (.versioned-repository? @ds-atom reponame) false) "Updated repo with versioning false should return false for versioned-repository? method")
 
       (.delete-repository @ds-atom reponame))))
+
+
+(deftest test-race-conditions
+  (testing "run multiple slow save operations and check if they were performed without exceptions."
+    (let [reponame (str (gensym "r"))
+          doc-id "rc-test"
+          base-doc (generate-document)
+          random-data-fn #(shelper/get-random-code  50000000)
+          parallel-count 25]
+      (.create-repository @ds-atom {:name reponame
+                                    :caching false
+                                    :versioning true})
+      (try
+        (println "Save basedoc")
+        (.save-document @ds-atom doc-id base-doc reponame)
+        (println "Save basedoc finished")
+        (let [promises (repeatedly parallel-count promise)
+              random-data (random-data-fn)
+              futures (for [v (range parallel-count)]
+                        (future
+                          (let [changed-doc (assoc base-doc
+                                                   :random-data random-data
+                                                   :parallel-version v)]
+                            (println (str "Start Parallel Doc Save " v))
+                            (.save-document @ds-atom doc-id changed-doc reponame)
+                            (println (str "End Parallel Doc Save " v))
+                            (deliver (nth promises v) true))))]
+          (doseq [v (range parallel-count)]
+            (deref (nth futures v)))
+          (doseq [v (range parallel-count)]
+            (is (true? (deref (nth promises v))) "Every Save should deliver true value")))
+        (catch Exception e
+          (is false (str "Exception in race-condition test: " e)))
+        (finally
+          (.delete-repository @ds-atom reponame))))))
+