git.iko.soy/typeable/bloodhound
on master help sign in
positive validation for bulk API
commitb8ca2fd69eda57fbdf1d21216844ea2fd19dd5ac authorChris Allen <cma@bitemyapp.com> dateSat 12 Apr 2014, 02:56 (13 years ago) parent6b50fdc
2 files changed+36−6
compare with master
unifiedsplit
Database/Bloodhound/Client.hs+13−6 tests/tests.hs+23
Database/Bloodhound/Client.hs+13−6
@@ -15,6 +15,7 @@ module Database.Bloodhound.Client
1515 , searchByType
1616 , refreshIndex
1717 , mkSearch
18+ , bulk
1819 , IndexSettings(..)
1920 , Server(..)
2021 , Reply(..)
@@ -54,6 +55,7 @@ module Database.Bloodhound.Client
5455 , DocId(..)
5556 , CacheName(..)
5657 , CacheKey(..)
58+ , BulkOperation(..)
5759 )
5860 where
5961
@@ -63,7 +65,7 @@ import Data.Aeson
6365import Data.Aeson.TH (deriveJSON)
6466import qualified Data.ByteString.Lazy.Char8 as L
6567import Data.ByteString.Builder
66−import Data.List (intercalate)
68+import Data.List (foldl', intercalate, intersperse)
6769import Data.Maybe (fromMaybe)
6870import Data.Monoid
6971import Data.Text (Text)
@@ -267,8 +269,16 @@ bulk :: Server -> [BulkOperation] -> IO Reply
267269bulk (Server server) bulkOps = dispatch url method body where
268270 url = joinPath [server, "_bulk"]
269271 method = NHTM.methodPost
270− blobs = concat $ fmap getStreamChunk bulkOps
271− body = Just $ toLazyByteString $ mash (mempty :: Builder) blobs
272+ body = Just $ collapseStream bulkOps
273+
274+collapseStream :: [BulkOperation] -> L.ByteString
275+collapseStream stream = collapsed where
276+ blobs = intersperse "\n" $ concat $ fmap getStreamChunk stream
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
272282
273283data BulkOperation =
274284 BulkIndex IndexName MappingName DocId Value
@@ -276,9 +286,6 @@ data BulkOperation =
276286 | BulkDelete IndexName MappingName DocId
277287 | BulkUpdate IndexName MappingName DocId Value deriving (Eq, Show)
278288
279−mash :: Builder -> [L.ByteString] -> Builder
280−mash builder xs = foldr (\x b -> mappend b (lazyByteString x)) builder xs
281−
282289mkMetadataValue :: Text -> String -> String -> String -> Value
283290mkMetadataValue operation indexName mappingName docId =
284291 object [operation .=
tests/tests.hs+23
@@ -81,6 +81,10 @@ searchExpectNoResults search = do
8181 let emptyHits = fmap (hits . searchHits) result
8282 emptyHits `shouldBe` Right []
8383
84+data BulkTest = BulkTest { name :: Text } deriving (Eq, Generic, Show)
85+instance FromJSON BulkTest
86+instance ToJSON BulkTest
87+
8488main :: IO ()
8589main = hspec $ do
8690
@@ -101,6 +105,25 @@ main = hspec $ do
101105 (responseBody docInserted) :: Either String (EsResult Tweet)
102106 fmap _source newTweet `shouldBe` Right exampleTweet
103107
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+
104127 describe "query API" $ do
105128 it "returns document for term query and identity filter" $ do
106129 _ <- insertData