git.iko.soy/typeable/bloodhound
on master help sign in
refactored bulk operations encode and added encodeBulkOperation(s)
commitb911e9adc23302d224cd5edb8f42803b28822268 authorChris Allen <cma@bitemyapp.com> dateSat 8 Nov 2014, 01:53 (12 years ago) parent1ea5fb8
4 files changed+76−35
compare with master
unifiedsplit
.travis.yml+6−5 Database/Bloodhound/Client.hs+32−28 README.org+37−1 tests/tests.hs+1−1
.travis.yml+6−5
@@ -7,11 +7,12 @@
77# release of a major GHC version. Setting HPVER implictly sets
88# GHCVER. Omit lines with versions you don't need/want testing for.
99env:
10− - GHCVER=7.6.3 ESVER=1.3.2
11− - GHCVER=7.8.2 ESVER=1.0.3
12− - GHCVER=7.8.2 ESVER=1.1.2
13− - GHCVER=7.8.2 ESVER=1.2.3
14− - GHCVER=7.8.2 ESVER=1.3.2
10+ - GHCVER=7.6.3 ESVER=1.4.0
11+ - GHCVER=7.8.3 ESVER=1.0.3
12+ - GHCVER=7.8.3 ESVER=1.1.2
13+ - GHCVER=7.8.3 ESVER=1.2.3
14+ - GHCVER=7.8.3 ESVER=1.3.2
15+ - GHCVER=7.8.3 ESVER=1.4.0
1516
1617# services:
1718# - elasticsearch
Database/Bloodhound/Client.hs+32−28
@@ -21,15 +21,18 @@ module Database.Bloodhound.Client
2121 , mkShardCount
2222 , mkReplicaCount
2323 , getStatus
24+ , encodeBulkOperations
25+ , encodeBulkOperation
2426 )
2527 where
2628
2729import Data.Aeson
2830import qualified Data.ByteString.Lazy.Char8 as L
2931import Data.ByteString.Builder
30−import Data.List (foldl', intercalate, intersperse)
32+import Data.List (intercalate)
3133import Data.Maybe (fromMaybe)
3234import Data.Text (Text)
35+import qualified Data.Vector as V
3336import Network.HTTP.Client
3437import qualified Network.HTTP.Types.Method as NHTM
3538import qualified Network.HTTP.Types.Status as NHTS
@@ -167,52 +170,53 @@ deleteDocument (Server server) (IndexName indexName)
167170 (MappingName mappingName) (DocId docId) =
168171 delete $ joinPath [server, indexName, mappingName, docId]
169172
170−bulk :: Server -> [BulkOperation] -> IO Reply
173+bulk :: Server -> V.Vector BulkOperation -> IO Reply
171174bulk (Server server) bulkOps = post url body where
172175 url = joinPath [server, "_bulk"]
173− body = Just $ collapseStream bulkOps
176+ body = Just $ encodeBulkOperations bulkOps
174177
175−collapseStream :: [BulkOperation] -> L.ByteString
176−collapseStream stream = collapsed where
177− blobs = intersperse "\n" $ concat $ fmap getStreamChunk stream
178+encodeBulkOperations :: V.Vector BulkOperation -> L.ByteString
179+encodeBulkOperations stream = collapsed where
180+ blobs = fmap encodeBulkOperation stream
178181 mashedTaters = mash (mempty :: Builder) blobs
179182 collapsed = toLazyByteString $ mappend mashedTaters (byteString "\n")
180183
181−mash :: Builder -> [L.ByteString] -> Builder
182−mash = foldl' (\b x -> mappend b (lazyByteString x))
184+mash :: Builder -> V.Vector L.ByteString -> Builder
185+mash = V.foldl' (\b x -> b `mappend` "\n" `mappend` (lazyByteString x))
183186
184−mkMetadataValue :: Text -> String -> String -> String -> Value
185−mkMetadataValue operation indexName mappingName docId =
187+mkBulkStreamValue :: Text -> String -> String -> String -> Value
188+mkBulkStreamValue operation indexName mappingName docId =
186189 object [operation .=
187190 object [ "_index" .= indexName
188191 , "_type" .= mappingName
189192 , "_id" .= docId]]
190193
191−getStreamChunk :: BulkOperation -> [L.ByteString]
192−getStreamChunk (BulkIndex (IndexName indexName)
194+encodeBulkOperation :: BulkOperation -> L.ByteString
195+encodeBulkOperation (BulkIndex (IndexName indexName)
193196 (MappingName mappingName)
194− (DocId docId) value) = blob where
195− metadata = mkMetadataValue "index" indexName mappingName docId
196− blob = [encode metadata, encode value]
197+ (DocId docId) value) = blob
198+ where metadata = mkBulkStreamValue "index" indexName mappingName docId
199+ blob = encode metadata `mappend` "\n" `mappend` encode value
197200
198−getStreamChunk (BulkCreate (IndexName indexName)
201+encodeBulkOperation (BulkCreate (IndexName indexName)
199202 (MappingName mappingName)
200− (DocId docId) value) = blob where
201− metadata = mkMetadataValue "create" indexName mappingName docId
202− blob = [encode metadata, encode value]
203+ (DocId docId) value) = blob
204+ where metadata = mkBulkStreamValue "create" indexName mappingName docId
205+ blob = encode metadata `mappend` "\n" `mappend` encode value
203206
204−getStreamChunk (BulkDelete (IndexName indexName)
207+encodeBulkOperation (BulkDelete (IndexName indexName)
205208 (MappingName mappingName)
206− (DocId docId)) = blob where
207− metadata = mkMetadataValue "delete" indexName mappingName docId
208− blob = [encode metadata]
209+ (DocId docId)) = blob
210+ where metadata = mkBulkStreamValue "delete" indexName mappingName docId
211+ blob = encode metadata
209212
210−getStreamChunk (BulkUpdate (IndexName indexName)
213+encodeBulkOperation (BulkUpdate (IndexName indexName)
211214 (MappingName mappingName)
212− (DocId docId) value) = blob where
213− metadata = mkMetadataValue "update" indexName mappingName docId
214− doc = object ["doc" .= value]
215− blob = [encode metadata, encode doc]
215+ (DocId docId) value) = blob
216+ where metadata = mkBulkStreamValue "update" indexName mappingName docId
217+ doc = object ["doc" .= value]
218+ blob = encode metadata `mappend` "\n" `mappend` encode doc
219+
216220
217221getDocument :: Server -> IndexName -> MappingName
218222 -> DocId -> IO Reply
README.org+37−1
@@ -45,6 +45,7 @@ http://hackage.haskell.org/package/bloodhound
4545-- the multi-line examples into your ghci REPL.
4646
4747:set -XDeriveGeneric
48+:{
4849import Database.Bloodhound
4950import Data.Aeson
5051import Data.Either (Either(..))
@@ -68,6 +69,7 @@ let defaultIndexSettings = IndexSettings (ShardCount 3) (ReplicaCount 2)
6869
6970-- response :: Reply, Reply is a synonym for Network.HTTP.Conduit.Response
7071response <- createIndex testServer defaultIndexSettings testIndex
72+:}
7173
7274#+END_SRC
7375
@@ -281,7 +283,7 @@ Right (Tweet {user = "bitemyapp"
281283
282284*** Bulk create, index
283285
284−#+BEGIN_SRC haskell
286+#+BEGIN_SRC haskell
285287
286288-- don't forget the imports and derive generic setting for ghci
287289-- at the beginning of the examples.
@@ -318,6 +320,40 @@ resp <- bulk testServer stream
318320
319321#+END_SRC
320322
323+** Encoding individual bulk API operations
324+
325+#+BEGIN_SRC haskell
326+-- the following functions are exported in Bloodhound so
327+-- you can build up bulk operations yourself
328+encodeBulkOperations :: V.Vector BulkOperation -> L.ByteString
329+encodeBulkOperation :: BulkOperation -> L.ByteString
330+
331+-- How to use the above:
332+data BulkTest = BulkTest { name :: Text } deriving (Eq, Generic, Show)
333+instance FromJSON BulkTest
334+instance ToJSON BulkTest
335+
336+_ <- insertData
337+let firstTest = BulkTest "blah"
338+let secondTest = BulkTest "bloo"
339+let firstDoc = BulkIndex testIndex
340+ testMapping (DocId "2") (toJSON firstTest)
341+let secondDoc = BulkCreate testIndex
342+ testMapping (DocId "3") (toJSON secondTest)
343+let stream = V.fromList [firstDoc, secondDoc] :: V.Vector BulkOperation
344+
345+-- to encode yourself
346+let firstDocEncoded = encode firstDoc :: L.ByteString
347+
348+-- to encode a vector of bulk operations
349+let encodedOperations = encodeBulkOperations stream
350+
351+-- to insert into a particular server
352+-- bulk :: Server -> V.Vector BulkOperation -> IO Reply
353+_ <- bulk testServer stream
354+
355+#+END_SRC
356+
321357** Search
322358
323359*** Querying
tests/tests.hs+1−1
@@ -240,7 +240,7 @@ main = hspec $ do
240240 testMapping (DocId "2") (toJSON firstTest)
241241 let secondDoc = BulkCreate testIndex
242242 testMapping (DocId "3") (toJSON secondTest)
243− let stream = [firstDoc, secondDoc]
243+ let stream = V.fromList [firstDoc, secondDoc]
244244 _ <- bulk testServer stream
245245 _ <- refreshIndex testServer testIndex
246246 fDoc <- getDocument testServer testIndex testMapping (DocId "2")