git.iko.soy/typeable/bloodhound
on master help sign in
positive validation for bulk API
b8ca2fdChris Allen
Database/Bloodhound/Client.hs+13−6
compare with master
unifiedsplit
@@ -15,6 +15,7 @@ module Database.Bloodhound.Client
15 , searchByType
15 , searchByType
16 , refreshIndex
16 , refreshIndex
17 , mkSearch
17 , mkSearch
18+ , bulk
18 , IndexSettings(..)
19 , IndexSettings(..)
19 , Server(..)
20 , Server(..)
20 , Reply(..)
21 , Reply(..)
@@ -54,6 +55,7 @@ module Database.Bloodhound.Client
54 , DocId(..)
55 , DocId(..)
55 , CacheName(..)
56 , CacheName(..)
56 , CacheKey(..)
57 , CacheKey(..)
58+ , BulkOperation(..)
57 )
59 )
58 where
60 where
59
61
@@ -63,7 +65,7 @@ import Data.Aeson
63import Data.Aeson.TH (deriveJSON)
65import Data.Aeson.TH (deriveJSON)
64import qualified Data.ByteString.Lazy.Char8 as L
66import qualified Data.ByteString.Lazy.Char8 as L
65import Data.ByteString.Builder
67import Data.ByteString.Builder
66−import Data.List (intercalate)
68+import Data.List (foldl', intercalate, intersperse)
67import Data.Maybe (fromMaybe)
69import Data.Maybe (fromMaybe)
68import Data.Monoid
70import Data.Monoid
69import Data.Text (Text)
71import Data.Text (Text)
@@ -267,8 +269,16 @@ bulk :: Server -> [BulkOperation] -> IO Reply
267bulk (Server server) bulkOps = dispatch url method body where
269bulk (Server server) bulkOps = dispatch url method body where
268 url = joinPath [server, "_bulk"]
270 url = joinPath [server, "_bulk"]
269 method = NHTM.methodPost
271 method = NHTM.methodPost
272+ body = Just $ collapseStream bulkOps
273+
274+collapseStream :: [BulkOperation] -> L.ByteString
275+collapseStream stream = collapsed where
270− blobs = concat $ fmap getStreamChunk bulkOps
276+ blobs = intersperse "\n" $ concat $ fmap getStreamChunk stream
271− body = Just $ toLazyByteString $ mash (mempty :: Builder) blobs
277+ mashedTaters = mash (mempty :: Builder) blobs
278+ collapsed = toLazyByteString $ mappend mashedTaters "\n"
279+
280+mash :: Builder -> [L.ByteString] -> Builder
281+mash builder xs = foldl' (\b x -> mappend b (lazyByteString x)) builder xs
272
282
273data BulkOperation =
283data BulkOperation =
274 BulkIndex IndexName MappingName DocId Value
284 BulkIndex IndexName MappingName DocId Value
@@ -276,9 +286,6 @@ data BulkOperation =
276 | BulkDelete IndexName MappingName DocId
286 | BulkDelete IndexName MappingName DocId
277 | BulkUpdate IndexName MappingName DocId Value deriving (Eq, Show)
287 | BulkUpdate IndexName MappingName DocId Value deriving (Eq, Show)
278
288
279−mash :: Builder -> [L.ByteString] -> Builder
280−mash builder xs = foldr (\x b -> mappend b (lazyByteString x)) builder xs
281−
282mkMetadataValue :: Text -> String -> String -> String -> Value
289mkMetadataValue :: Text -> String -> String -> String -> Value
283mkMetadataValue operation indexName mappingName docId =
290mkMetadataValue operation indexName mappingName docId =
284 object [operation .=
291 object [operation .=
tests/tests.hs+23
@@ -81,6 +81,10 @@ searchExpectNoResults search = do
81 let emptyHits = fmap (hits . searchHits) result
81 let emptyHits = fmap (hits . searchHits) result
82 emptyHits `shouldBe` Right []
82 emptyHits `shouldBe` Right []
83
83
84+data BulkTest = BulkTest { name :: Text } deriving (Eq, Generic, Show)
85+instance FromJSON BulkTest
86+instance ToJSON BulkTest
87+
84main :: IO ()
88main :: IO ()
85main = hspec $ do
89main = hspec $ do
86
90
@@ -101,6 +105,25 @@ main = hspec $ do
101 (responseBody docInserted) :: Either String (EsResult Tweet)
105 (responseBody docInserted) :: Either String (EsResult Tweet)
102 fmap _source newTweet `shouldBe` Right exampleTweet
106 fmap _source newTweet `shouldBe` Right exampleTweet
103
107
108+ describe "bulk API" $ do
109+ it "inserts all documents we request" $ do
110+ _ <- insertData
111+ let firstTest = BulkTest "blah"
112+ let secondTest = BulkTest "bloo"
113+ let firstDoc = BulkIndex (IndexName "twitter")
114+ (MappingName "tweet") (DocId "2") (object ["name" .= String "blah"])
115+ let secondDoc = BulkCreate (IndexName "twitter")
116+ (MappingName "tweet") (DocId "3") (object ["name" .= String "bloo"])
117+ let stream = [firstDoc, secondDoc]
118+ response <- bulk testServer stream
119+ _ <- refreshIndex testServer testIndex
120+ fDoc <- getDocument testServer testIndex testMapping (DocId "2")
121+ sDoc <- getDocument testServer testIndex testMapping (DocId "3")
122+ let maybeFirst = eitherDecode $ responseBody fDoc :: Either String (EsResult BulkTest)
123+ let maybeSecond = eitherDecode $ responseBody sDoc :: Either String (EsResult BulkTest)
124+ fmap _source maybeFirst `shouldBe` Right firstTest
125+ fmap _source maybeSecond `shouldBe` Right secondTest
126+
104 describe "query API" $ do
127 describe "query API" $ do
105 it "returns document for term query and identity filter" $ do
128 it "returns document for term query and identity filter" $ do
106 _ <- insertData
129 _ <- insertData