404d807 · typeable/bloodhound

git.iko.soy/typeable/bloodhound

on master help sign in

Merge pull request #45 from Soostone/44-versioning

404d807Chris Allen11 years ago

README.md+2−2

compare with master

unifiedsplit

@@ -225,7 +225,7 @@ instance FromJSON Location

225-- λ> encode $ Location 10.0 10.0

226-- "{\"lat\":10,\"lon\":10}"

227

228−resp <- withBH' $ indexDocument testIndex testMapping exampleTweet (DocId "1")

228+resp <- withBH' $ indexDocument testIndex testMapping defaultIndexDocumentSettings exampleTweet (DocId "1")

229

230```

231

@@ -368,7 +368,7 @@ let encodedOperations = encodeBulkOperations stream

368

369-- to insert into a particular server

370-- bulk :: V.Vector BulkOperation -> IO Reply

371−_ <- withBH' $ bulk stream

371+_ <- withBH' $ bulk streamp

372

373```

374

bloodhound.cabal+2−1

@@ -37,7 +37,8 @@ library

37 vector >= 0.10.9 && <0.11,

38 uri-bytestring >= 0.1 && <0.2,

39 exceptions,

40− data-default-class

40+ data-default-class,

41+ blaze-builder

41 default-language: Haskell2010

42 default-language: Haskell2010

43test-suite tests

44test-suite tests

src/Database/Bloodhound/Client.hs+46−4

@@ -47,9 +47,14 @@ module Database.Bloodhound.Client

47 , getStatus

48 , encodeBulkOperations

49 , encodeBulkOperation

50+ -- * Reply-handling tools

51+ , isVersionConflict

52+ , isSuccess

53+ , isCreated

50 )

54 )

51 where

55 where

57+import qualified Blaze.ByteString.Builder as BB

53import Control.Applicative

58import Control.Applicative

54import Control.Monad

59import Control.Monad

55import Control.Monad.Catch

60import Control.Monad.Catch

@@ -58,7 +63,9 @@ import Data.Aeson

58import Data.ByteString.Lazy.Builder

63import Data.ByteString.Lazy.Builder

59import qualified Data.ByteString.Lazy.Char8 as L

64import qualified Data.ByteString.Lazy.Char8 as L

60import Data.Default.Class

65import Data.Default.Class

66+import Data.Ix

61import Data.Maybe (fromMaybe)

67import Data.Maybe (fromMaybe)

68+import Data.Monoid

62import Data.Text (Text)

69import Data.Text (Text)

63import qualified Data.Text as T

70import qualified Data.Text as T

64import qualified Data.Text.Encoding as T

71import qualified Data.Text.Encoding as T

@@ -66,6 +73,7 @@ import qualified Data.Vector as V

66import Network.HTTP.Client

73import Network.HTTP.Client

67import qualified Network.HTTP.Types.Method as NHTM

74import qualified Network.HTTP.Types.Method as NHTM

68import qualified Network.HTTP.Types.Status as NHTS

75import qualified Network.HTTP.Types.Status as NHTS

76+import qualified Network.HTTP.Types.URI as NHTU

69import Prelude hiding (filter, head)

77import Prelude hiding (filter, head)

70import URI.ByteString hiding (Query)

78import URI.ByteString hiding (Query)

@@ -160,6 +168,16 @@ joinPath ps = do

160 Server s <- bhServer <$> getBHEnv

168 Server s <- bhServer <$> getBHEnv

161 return $ joinPath' (s:ps)

169 return $ joinPath' (s:ps)

162

170

171+-- | Severely dumbed down query renderer. Assumes your data doesn't

172+-- need any encoding

173+addQuery :: [(Text, Maybe Text)] -> Text -> Text

174+addQuery q u = u <> rendered

175+ where

176+ rendered =

177+ T.decodeUtf8 $ BB.toByteString $ NHTU.renderQueryText prependQuestionMark q

178+ prependQuestionMark = True

179+

180+

163bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c

181bindM2 :: (Applicative m, Monad m) => (a -> b -> m c) -> m a -> m b -> m c

164bindM2 f ma mb = join (f <$> ma <*> mb)

182bindM2 f ma mb = join (f <$> ma <*> mb)

165

183

@@ -315,15 +333,25 @@ deleteMapping (IndexName indexName)

315-- convert into a JSON 'Value'. The 'DocId' will function as the

333-- convert into a JSON 'Value'. The 'DocId' will function as the

316-- primary key for the document.

334-- primary key for the document.

317--

335--

318−-- >>> resp <- runBH' $ indexDocument testIndex testMapping exampleTweet (DocId "1")

336+-- >>> resp <- runBH' $ indexDocument testIndex testMapping defaultIndexDocumentSettings exampleTweet (DocId "1")

319-- >>> print resp

337-- >>> print resp

320-- 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}

338-- 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}

321indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName

339indexDocument :: (ToJSON doc, MonadBH m) => IndexName -> MappingName

322− -> doc -> DocId -> m Reply

340+ -> IndexDocumentSettings -> doc -> DocId -> m Reply

323indexDocument (IndexName indexName)

341indexDocument (IndexName indexName)

324− (MappingName mappingName) document (DocId docId) =

342+ (MappingName mappingName) cfg document (DocId docId) =

325 bindM2 put url (return body)

343 bindM2 put url (return body)

326− where url = joinPath [indexName, mappingName, docId]

344+ where url = addQuery params <$> joinPath [indexName, mappingName, docId]

345+ params = case idsVersionControl cfg of

346+ NoVersionControl -> []

347+ InternalVersion v -> versionParams v "internal"

348+ ExternalGT (ExternalDocVersion v) -> versionParams v "external_gt"

349+ ExternalGTE (ExternalDocVersion v) -> versionParams v "external_gte"

350+ ForceVersion (ExternalDocVersion v) -> versionParams v "force"

351+ vt = T.pack . show . docVersionNumber

352+ versionParams v t = [ ("version", Just $ vt v)

353+ , ("version_type", Just t)

354+ ]

327 body = Just (encode document)

355 body = Just (encode document)

328

356

329-- | 'deleteDocument' is the primary way to delete a single document.

357-- | 'deleteDocument' is the primary way to delete a single document.

@@ -536,3 +564,17 @@ setURI req URI{..} = do

536 Scheme "https" -> True

564 Scheme "https" -> True

537 _ -> False

565 _ -> False

538 theQueryString = [(k , Just v) | (k, v) <- queryPairs uriQuery]

566 theQueryString = [(k , Just v) | (k, v) <- queryPairs uriQuery]

567+

568+-- | Was there an optimistic concurrency control conflict when

569+-- indexing a document?

570+isVersionConflict :: Reply -> Bool

571+isVersionConflict = statusCheck (== 409)

572+

573+isSuccess :: Reply -> Bool

574+isSuccess = statusCheck (inRange (200, 299))

575+

576+isCreated :: Reply -> Bool

577+isCreated = statusCheck (== 201)

578+

579+statusCheck :: (Int -> Bool) -> Reply -> Bool

580+statusCheck prd = prd . NHTS.statusCode . responseStatus

src/Database/Bloodhound/Types.hs+101−1

@@ -29,6 +29,7 @@

29module Database.Bloodhound.Types

30 ( defaultCache

31 , defaultIndexSettings

32+ , defaultIndexDocumentSettings

32 , mkSort

33 , mkSort

33 , showText

34 , showText

34 , unpackId

35 , unpackId

@@ -41,6 +42,8 @@ module Database.Bloodhound.Types

41 , mkTermsAggregation

42 , mkTermsAggregation

42 , mkTermsScriptAggregation

43 , mkTermsScriptAggregation

43 , mkDateHistogram

44 , mkDateHistogram

45+ , mkDocVersion

46+ , docVersionNumber

44 , toTerms

47 , toTerms

45 , toDateHistogram

48 , toDateHistogram

46 , omitNulls

49 , omitNulls

@@ -56,6 +59,10 @@ module Database.Bloodhound.Types

56 , Server(..)

59 , Server(..)

57 , Reply

60 , Reply

58 , EsResult(..)

61 , EsResult(..)

62+ , DocVersion

63+ , ExternalDocVersion(..)

64+ , VersionControl(..)

65+ , IndexDocumentSettings(..)

59 , Query(..)

66 , Query(..)

60 , Search(..)

67 , Search(..)

61 , SearchResult(..)

68 , SearchResult(..)

@@ -218,10 +225,12 @@ import qualified Data.ByteString.Lazy.Char8 as L

218import Data.List (nub)

225import Data.List (nub)

219import Data.List.NonEmpty (NonEmpty (..), toList)

226import Data.List.NonEmpty (NonEmpty (..), toList)

220import qualified Data.Map.Strict as M

227import qualified Data.Map.Strict as M

228+import Data.Maybe

221import Data.Text (Text)

229import Data.Text (Text)

222import qualified Data.Text as T

230import qualified Data.Text as T

223import Data.Time.Clock (UTCTime)

231import Data.Time.Clock (UTCTime)

224import qualified Data.Vector as V

232import qualified Data.Vector as V

233+import GHC.Enum

225import GHC.Generics (Generic)

234import GHC.Generics (Generic)

226import Network.HTTP.Client

235import Network.HTTP.Client

227import qualified Network.HTTP.Types.Method as NHTM

236import qualified Network.HTTP.Types.Method as NHTM

@@ -371,10 +380,78 @@ data BulkOperation =

371data EsResult a = EsResult { _index :: Text

380data EsResult a = EsResult { _index :: Text

372 , _type :: Text

381 , _type :: Text

373 , _id :: Text

382 , _id :: Text

374− , _version :: Int

383+ , _version :: DocVersion

375 , found :: Maybe Bool

384 , found :: Maybe Bool

376 , _source :: a } deriving (Eq, Show)

385 , _source :: a } deriving (Eq, Show)

377

386

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+

378{-| 'Sort' is a synonym for a list of 'SortSpec's. Sort behavior is order

455{-| 'Sort' is a synonym for a list of 'SortSpec's. Sort behavior is order

379 dependent with later sorts acting as tie-breakers for earlier sorts.

456 dependent with later sorts acting as tie-breakers for earlier sorts.

380-}

457-}

@@ -2173,3 +2250,26 @@ instance FromJSON ShardResult where

2173 v .: "successful" <*>

2250 v .: "successful" <*>

2174 v .: "failed"

2251 v .: "failed"

2175 parseJSON _ = empty

2252 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+43−6

@@ -5,6 +5,7 @@

5module Main where

7import Control.Applicative

8+import Control.Exception

8import Control.Monad

9import Control.Monad

9import Control.Monad.Reader

10import Control.Monad.Reader

10import Data.Aeson

11import Data.Aeson

@@ -25,7 +26,7 @@ import Network.HTTP.Client

25import qualified Network.HTTP.Types.Status as NHTS

26import qualified Network.HTTP.Types.Status as NHTS

26import Prelude hiding (filter)

27import Prelude hiding (filter)

27import Test.Hspec

28import Test.Hspec

28−import Test.QuickCheck.Property.Monoid

29+import Test.QuickCheck.Property.Monoid (prop_Monoid, eq, T(..))

30import Test.Hspec.QuickCheck (prop)

31import Test.Hspec.QuickCheck (prop)

31import Test.QuickCheck

32import Test.QuickCheck

@@ -147,18 +148,27 @@ otherTweet = Tweet { user = "notmyapp"

147 , age = 1000

148 , age = 1000

148 , location = Location 40.12 (-71.34) }

149 , location = Location 40.12 (-71.34) }

149

150

150−insertData :: BH IO ()

151+resetIndex :: BH IO ()

151−insertData = do

152+resetIndex = do

152 _ <- deleteExampleIndex

153 _ <- deleteExampleIndex

153 _ <- createExampleIndex

154 _ <- createExampleIndex

154 _ <- putMapping testIndex testMapping TweetMapping

155 _ <- putMapping testIndex testMapping TweetMapping

155− _ <- indexDocument testIndex testMapping exampleTweet (DocId "1")

156− _ <- refreshIndex testIndex

157 return ()

156 return ()

158

157

158+insertData :: BH IO Reply

159+insertData = do

160+ resetIndex

161+ insertData' defaultIndexDocumentSettings

162+

163+insertData' :: IndexDocumentSettings -> BH IO Reply

164+insertData' ids = do

165+ r <- indexDocument testIndex testMapping ids exampleTweet (DocId "1")

166+ _ <- refreshIndex testIndex

167+ return r

168+

159insertOther :: BH IO ()

169insertOther :: BH IO ()

160insertOther = do

170insertOther = do

161− _ <- indexDocument testIndex testMapping otherTweet (DocId "2")

171+ _ <- indexDocument testIndex testMapping defaultIndexDocumentSettings otherTweet (DocId "2")

162 _ <- refreshIndex testIndex

172 _ <- refreshIndex testIndex

163 return ()

173 return ()

164

174

@@ -290,6 +300,14 @@ main = hspec $ do

290 (responseBody docInserted) :: Either String (EsResult Tweet)

300 (responseBody docInserted) :: Either String (EsResult Tweet)

291 liftIO $ (fmap _source newTweet `shouldBe` Right exampleTweet)

301 liftIO $ (fmap _source newTweet `shouldBe` Right exampleTweet)

292

302

303+ it "can use optimistic concurrency control" $ withTestEnv $ do

304+ let ev = ExternalDocVersion minBound

305+ let cfg = defaultIndexDocumentSettings { idsVersionControl = ExternalGT ev }

306+ resetIndex

307+ res <- insertData' cfg

308+ liftIO $ isCreated res `shouldBe` True

309+ res' <- insertData' cfg

310+ liftIO $ isVersionConflict res' `shouldBe` True

293

311

294 describe "bulk API" $ do

312 describe "bulk API" $ do

295 it "inserts all documents we request" $ withTestEnv $ do

313 it "inserts all documents we request" $ withTestEnv $ do

@@ -659,3 +677,22 @@ main = hspec $ do

659 describe "Monoid (SearchHits a)" $ do

677 describe "Monoid (SearchHits a)" $ do

660 prop "abides the monoid laws" $ eq $

678 prop "abides the monoid laws" $ eq $

661 prop_Monoid (T :: T (SearchHits ()))

679 prop_Monoid (T :: T (SearchHits ()))

680+

681+ describe "mkDocVersion" $ do

682+ prop "can never construct an out of range docVersion" $ \i ->

683+ let res = mkDocVersion i

684+ in case res of

685+ Nothing -> property True

686+ Just dv -> (dv >= minBound) .&&.

687+ (dv <= maxBound) .&&.

688+ docVersionNumber dv === i

689+

690+ describe "Enum DocVersion" $ do

691+ it "follows the laws of Enum, Bounded" $ do

692+ evaluate (succ maxBound :: DocVersion) `shouldThrow` anyErrorCall

693+ evaluate (pred minBound :: DocVersion) `shouldThrow` anyErrorCall

694+ evaluate (toEnum 0 :: DocVersion) `shouldThrow` anyErrorCall

695+ evaluate (toEnum 9200000000000000001 :: DocVersion) `shouldThrow` anyErrorCall

696+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]

697+ enumFrom (pred maxBound :: DocVersion) `shouldBe` [pred maxBound, maxBound]

698+ enumFromThen minBound (pred maxBound :: DocVersion) `shouldBe` [minBound, pred maxBound]