git.iko.soy/typeable/bloodhound
on master help sign in
Add optimistic concurrency control version feature

This is a first attempt at addressing #44

commit2d8be0d3e7588653b043b6f0c8485db8d17f8c8c authorMichael Xavier <michael@michaelxavier.net> dateFri 10 Jul 2015, 11:29 (11 years ago) parentc7e0412
3 files changed+184−11
compare with master
unifiedsplit
src/Database/Bloodhound/Client.hs+41−4 src/Database/Bloodhound/Types.hs+101−1 tests/tests.hs+42−6
src/Database/Bloodhound/Client.hs+41−4
@@ -47,6 +47,10 @@ module Database.Bloodhound.Client
4747 , getStatus
4848 , encodeBulkOperations
4949 , encodeBulkOperation
50+ -- * Reply-handling tools
51+ , isVersionConflict
52+ , isSuccess
53+ , isCreated
5054 )
5155 where
5256
@@ -58,7 +62,9 @@ import Data.Aeson
5862import Data.ByteString.Lazy.Builder
5963import qualified Data.ByteString.Lazy.Char8 as L
6064import Data.Default.Class
65+import Data.Ix
6166import Data.Maybe (fromMaybe)
67+import Data.Monoid
6268import Data.Text (Text)
6369import qualified Data.Text as T
6470import qualified Data.Text.Encoding as T
@@ -160,6 +166,15 @@ joinPath ps = do
160166 Server s <- bhServer <$> getBHEnv
161167 return $ joinPath' (s:ps)
162168
169+-- | Severely dumbed down query renderer. Assumes your data doesn't
170+-- need any encoding
171+addQuery :: [(Text, Text)] -> Text -> Text
172+addQuery ps u = u <> "?" <> params
173+ where
174+ params = T.intercalate "&" (uncurry mkParam <$> ps)
175+ mkParam k v = k <> "=" <> v
176+
177+
163178bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c
164179bindM2 f ma mb = join (f <$> ma <*> mb)
165180
@@ -315,15 +330,23 @@ deleteMapping (IndexName indexName)
315330-- convert into a JSON 'Value'. The 'DocId' will function as the
316331-- primary key for the document.
317332--
318−-- >>> resp <- runBH' $ indexDocument testIndex testMapping exampleTweet (DocId "1")
333+-- >>> resp <- runBH' $ indexDocument defaultIndexDocumentSettings testIndex testMapping exampleTweet (DocId "1")
319334-- >>> print resp
320335-- Response {responseStatus = Status {statusCode = 201, statusMessage = "Created"}, responseVersion = HTTP/1.1, responseHeaders = [("Content-Type","application/json; charset=UTF-8"),("Content-Length","74")], responseBody = "{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"1\",\"_version\":1,\"created\":true}", responseCookieJar = CJ {expose = []}, responseClose' = ResponseClose}
321336indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName
322− -> doc -> DocId -> m Reply
337+ -> IndexDocumentSettings -> doc -> DocId -> m Reply
323338indexDocument (IndexName indexName)
324− (MappingName mappingName) document (DocId docId) =
339+ (MappingName mappingName) cfg document (DocId docId) =
325340 bindM2 put url (return body)
326− where url = joinPath [indexName, mappingName, docId]
341+ where url = addQuery params <$> joinPath [indexName, mappingName, docId]
342+ params = case idsVersionControl cfg of
343+ NoVersionControl -> []
344+ InternalVersion v -> versionParams v "internal"
345+ ExternalGT (ExternalDocVersion v) -> versionParams v "external_gt"
346+ ExternalGTE (ExternalDocVersion v) -> versionParams v "external_gte"
347+ ForceVersion (ExternalDocVersion v) -> versionParams v "force"
348+ vt = T.pack . show . docVersionNumber
349+ versionParams v t = [("version", vt v), ("version_type", t)]
327350 body = Just (encode document)
328351
329352-- | 'deleteDocument' is the primary way to delete a single document.
@@ -536,3 +559,17 @@ setURI req URI{..} = do
536559 Scheme "https" -> True
537560 _ -> False
538561 theQueryString = [(k , Just v) | (k, v) <- queryPairs uriQuery]
562+
563+-- | Was there an optimistic concurrency control conflict when
564+-- indexing a document?
565+isVersionConflict :: Reply -> Bool
566+isVersionConflict = statusCheck (== 409)
567+
568+isSuccess :: Reply -> Bool
569+isSuccess = statusCheck (inRange (200, 299))
570+
571+isCreated :: Reply -> Bool
572+isCreated = statusCheck (== 201)
573+
574+statusCheck :: (Int -> Bool) -> Reply -> Bool
575+statusCheck prd = prd . NHTS.statusCode . responseStatus
src/Database/Bloodhound/Types.hs+101−1
@@ -29,6 +29,7 @@
2929module Database.Bloodhound.Types
3030 ( defaultCache
3131 , defaultIndexSettings
32+ , defaultIndexDocumentSettings
3233 , mkSort
3334 , showText
3435 , unpackId
@@ -41,6 +42,8 @@ module Database.Bloodhound.Types
4142 , mkTermsAggregation
4243 , mkTermsScriptAggregation
4344 , mkDateHistogram
45+ , mkDocVersion
46+ , docVersionNumber
4447 , toTerms
4548 , toDateHistogram
4649 , omitNulls
@@ -56,6 +59,10 @@ module Database.Bloodhound.Types
5659 , Server(..)
5760 , Reply
5861 , EsResult(..)
62+ , DocVersion
63+ , ExternalDocVersion(..)
64+ , VersionControl(..)
65+ , IndexDocumentSettings(..)
5966 , Query(..)
6067 , Search(..)
6168 , SearchResult(..)
@@ -218,10 +225,12 @@ import qualified Data.ByteString.Lazy.Char8 as L
218225import Data.List (nub)
219226import Data.List.NonEmpty (NonEmpty (..), toList)
220227import qualified Data.Map.Strict as M
228+import Data.Maybe
221229import Data.Text (Text)
222230import qualified Data.Text as T
223231import Data.Time.Clock (UTCTime)
224232import qualified Data.Vector as V
233+import GHC.Enum
225234import GHC.Generics (Generic)
226235import Network.HTTP.Client
227236import qualified Network.HTTP.Types.Method as NHTM
@@ -371,10 +380,78 @@ data BulkOperation =
371380data EsResult a = EsResult { _index :: Text
372381 , _type :: Text
373382 , _id :: Text
374− , _version :: Int
383+ , _version :: DocVersion
375384 , found :: Maybe Bool
376385 , _source :: a } deriving (Eq, Show)
377386
387+{-| 'DocVersion' is an integer version number for a document between 1
388+and 9.2e+18 used for <<https://www.elastic.co/guide/en/elasticsearch/guide/current/optimistic-concurrency-control.html optimistic concurrency control>>.
389+-}
390+newtype DocVersion = DocVersion {
391+ docVersionNumber :: Int
392+ } deriving (Eq, Show, Ord, ToJSON)
393+
394+-- | Smart constructor for in-range doc version
395+mkDocVersion :: Int -> Maybe DocVersion
396+mkDocVersion i
397+ | i >= (docVersionNumber minBound) && i <= (docVersionNumber maxBound) =
398+ Just $ DocVersion i
399+ | otherwise = Nothing
400+
401+
402+{-| 'ExternalDocVersion' is a convenience wrapper if your code uses its
403+own version numbers instead of ones from ES.
404+-}
405+newtype ExternalDocVersion = ExternalDocVersion DocVersion
406+ deriving (Eq, Show, Ord, Bounded, Enum, ToJSON)
407+
408+{-| 'VersionControl' is specified when indexing documents as a
409+optimistic concurrency control.
410+-}
411+data VersionControl = NoVersionControl
412+ -- ^ Don't send a version. This is a pure overwrite.
413+ | InternalVersion DocVersion
414+ -- ^ Use the default ES versioning scheme. Only
415+ -- index the document if the version is the same
416+ -- as the one specified. Only applicable to
417+ -- updates, as you should be getting Version from
418+ -- a search result.
419+ | ExternalGT ExternalDocVersion
420+ -- ^ Use your own version numbering. Only index
421+ -- the document if the version is strictly higher
422+ -- OR the document doesn't exist. The given
423+ -- version will be used as the new version number
424+ -- for the stored document. N.B. All updates must
425+ -- increment this number, meaning there is some
426+ -- global, external ordering of updates.
427+ | ExternalGTE ExternalDocVersion
428+ -- ^ Use your own version numbering. Only index
429+ -- the document if the version is equal or higher
430+ -- than the stored version. Will succeed if there
431+ -- is no existing document. The given version will
432+ -- be used as the new version number for the
433+ -- stored document. Use with care, as this could
434+ -- result in data loss.
435+ | ForceVersion ExternalDocVersion
436+ -- ^ The document will always be indexed and the
437+ -- given version will be the new version. This is
438+ -- typically used for correcting errors. Use with
439+ -- care, as this could result in data loss.
440+ deriving (Show, Eq, Ord)
441+
442+{-| 'IndexDocumentSettings' are special settings supplied when indexing
443+a document. For the best backwards compatiblity when new fields are
444+added, you should probably prefer to start with 'defaultIndexDocumentSettings'
445+-}
446+data IndexDocumentSettings = IndexDocumentSettings {
447+ idsVersionControl :: VersionControl
448+ }
449+
450+{-| Reasonable default settings. Chooses no version control.
451+-}
452+defaultIndexDocumentSettings :: IndexDocumentSettings
453+defaultIndexDocumentSettings = IndexDocumentSettings NoVersionControl
454+
378455{-| 'Sort' is a synonym for a list of 'SortSpec's. Sort behavior is order
379456 dependent with later sorts acting as tie-breakers for earlier sorts.
380457-}
@@ -2173,3 +2250,26 @@ instance FromJSON ShardResult where
21732250 v .: "successful" <*>
21742251 v .: "failed"
21752252 parseJSON _ = empty
2253+
2254+
2255+instance FromJSON DocVersion where
2256+ parseJSON v = do
2257+ i <- parseJSON v
2258+ maybe (fail "DocVersion out of range") return $ mkDocVersion i
2259+
2260+instance Bounded DocVersion where
2261+ minBound = DocVersion 1
2262+ maxBound = DocVersion 9200000000000000000 -- 9.2e+18
2263+
2264+instance Enum DocVersion where
2265+ succ x
2266+ | x /= maxBound = DocVersion (succ $ docVersionNumber x)
2267+ | otherwise = succError "DocVersion"
2268+ pred x
2269+ | x /= minBound = DocVersion (pred $ docVersionNumber x)
2270+ | otherwise = predError "DocVersion"
2271+ toEnum i =
2272+ fromMaybe (error $ show i ++ " out of DocVersion range") $ mkDocVersion i
2273+ fromEnum = docVersionNumber
2274+ enumFrom = boundedEnumFrom
2275+ enumFromThen = boundedEnumFromThen
tests/tests.hs+42−6
@@ -25,7 +25,7 @@ import Network.HTTP.Client
2525import qualified Network.HTTP.Types.Status as NHTS
2626import Prelude hiding (filter)
2727import Test.Hspec
28−import Test.QuickCheck.Property.Monoid
28+import Test.QuickCheck.Property.Monoid (prop_Monoid, eq, T(..))
2929
3030import Test.Hspec.QuickCheck (prop)
3131import Test.QuickCheck
@@ -147,18 +147,27 @@ otherTweet = Tweet { user = "notmyapp"
147147 , age = 1000
148148 , location = Location 40.12 (-71.34) }
149149
150−insertData :: BH IO ()
151−insertData = do
150+resetIndex :: BH IO ()
151+resetIndex = do
152152 _ <- deleteExampleIndex
153153 _ <- createExampleIndex
154154 _ <- putMapping testIndex testMapping TweetMapping
155− _ <- indexDocument testIndex testMapping exampleTweet (DocId "1")
156− _ <- refreshIndex testIndex
157155 return ()
158156
157+insertData :: BH IO Reply
158+insertData = do
159+ resetIndex
160+ insertData' defaultIndexDocumentSettings
161+
162+insertData' :: IndexDocumentSettings -> BH IO Reply
163+insertData' ids = do
164+ r <- indexDocument testIndex testMapping ids exampleTweet (DocId "1")
165+ _ <- refreshIndex testIndex
166+ return r
167+
159168insertOther :: BH IO ()
160169insertOther = do
161− _ <- indexDocument testIndex testMapping otherTweet (DocId "2")
170+ _ <- indexDocument testIndex testMapping defaultIndexDocumentSettings otherTweet (DocId "2")
162171 _ <- refreshIndex testIndex
163172 return ()
164173
@@ -290,6 +299,14 @@ main = hspec $ do
290299 (responseBody docInserted) :: Either String (EsResult Tweet)
291300 liftIO $ (fmap _source newTweet `shouldBe` Right exampleTweet)
292301
302+ it "can use optimistic concurrency control" $ withTestEnv $ do
303+ let ev = ExternalDocVersion minBound
304+ let cfg = defaultIndexDocumentSettings { idsVersionControl = ExternalGT ev }
305+ resetIndex
306+ res <- insertData' cfg
307+ liftIO $ isCreated res `shouldBe` True
308+ res' <- insertData' cfg
309+ liftIO $ isVersionConflict res' `shouldBe` True
293310
294311 describe "bulk API" $ do
295312 it "inserts all documents we request" $ withTestEnv $ do
@@ -659,3 +676,22 @@ main = hspec $ do
659676 describe "Monoid (SearchHits a)" $ do
660677 prop "abides the monoid laws" $ eq $
661678 prop_Monoid (T :: T (SearchHits ()))
679+
680+ describe "mkDocVersion" $ do
681+ prop "can never construct an out of range docVersion" $ \i ->
682+ let res = mkDocVersion i
683+ in case res of
684+ Nothing -> property True
685+ Just dv -> (dv >= minBound) .&&.
686+ (dv <= maxBound) .&&.
687+ docVersionNumber dv === i
688+
689+ describe "Enum DocVersion" $ do
690+ it "follows the laws of Enum, Bounded" $ do
691+ return (succ maxBound :: DocVersion) `shouldThrow` anyErrorCall
692+ return (pred minBound :: DocVersion) `shouldThrow` anyErrorCall
693+ return (toEnum 0 :: DocVersion) `shouldThrow` anyErrorCall
694+ return (toEnum 9200000000000000001 :: DocVersion) `shouldThrow` anyErrorCall
695+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]
696+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]
697+ enumFromThen minBound (pred maxBound :: DocVersion) `shouldBe` [minBound, pred maxBound, maxBound]