git.iko.soy/typeable/bloodhound
on master help sign in
refactored bulk operations encode and added encodeBulkOperation(s)
b911e9aChris Allen
.travis.yml+6−5
compare with master
unifiedsplit
@@ -7,11 +7,12 @@
7# release of a major GHC version. Setting HPVER implictly sets
7# release of a major GHC version. Setting HPVER implictly sets
8# GHCVER. Omit lines with versions you don't need/want testing for.
8# GHCVER. Omit lines with versions you don't need/want testing for.
9env:
9env:
10− - GHCVER=7.6.3 ESVER=1.3.2
10+ - GHCVER=7.6.3 ESVER=1.4.0
11− - GHCVER=7.8.2 ESVER=1.0.3
11+ - GHCVER=7.8.3 ESVER=1.0.3
12− - GHCVER=7.8.2 ESVER=1.1.2
12+ - GHCVER=7.8.3 ESVER=1.1.2
13− - GHCVER=7.8.2 ESVER=1.2.3
13+ - GHCVER=7.8.3 ESVER=1.2.3
14− - GHCVER=7.8.2 ESVER=1.3.2
14+ - GHCVER=7.8.3 ESVER=1.3.2
15+ - GHCVER=7.8.3 ESVER=1.4.0
15
16
16# services:
17# services:
17# - elasticsearch
18# - elasticsearch
Database/Bloodhound/Client.hs+32−28
@@ -21,15 +21,18 @@ module Database.Bloodhound.Client
21 , mkShardCount
21 , mkShardCount
22 , mkReplicaCount
22 , mkReplicaCount
23 , getStatus
23 , getStatus
24+ , encodeBulkOperations
25+ , encodeBulkOperation
24 )
26 )
25 where
27 where
26
28
27import Data.Aeson
29import Data.Aeson
28import qualified Data.ByteString.Lazy.Char8 as L
30import qualified Data.ByteString.Lazy.Char8 as L
29import Data.ByteString.Builder
31import Data.ByteString.Builder
30−import Data.List (foldl', intercalate, intersperse)
32+import Data.List (intercalate)
31import Data.Maybe (fromMaybe)
33import Data.Maybe (fromMaybe)
32import Data.Text (Text)
34import Data.Text (Text)
35+import qualified Data.Vector as V
33import Network.HTTP.Client
36import Network.HTTP.Client
34import qualified Network.HTTP.Types.Method as NHTM
37import qualified Network.HTTP.Types.Method as NHTM
35import qualified Network.HTTP.Types.Status as NHTS
38import qualified Network.HTTP.Types.Status as NHTS
@@ -167,52 +170,53 @@ deleteDocument (Server server) (IndexName indexName)
167 (MappingName mappingName) (DocId docId) =
170 (MappingName mappingName) (DocId docId) =
168 delete $ joinPath [server, indexName, mappingName, docId]
171 delete $ joinPath [server, indexName, mappingName, docId]
169
172
170−bulk :: Server -> [BulkOperation] -> IO Reply
173+bulk :: Server -> V.Vector BulkOperation -> IO Reply
171bulk (Server server) bulkOps = post url body where
174bulk (Server server) bulkOps = post url body where
172 url = joinPath [server, "_bulk"]
175 url = joinPath [server, "_bulk"]
173− body = Just $ collapseStream bulkOps
176+ body = Just $ encodeBulkOperations bulkOps
174
177
175−collapseStream :: [BulkOperation] -> L.ByteString
178+encodeBulkOperations :: V.Vector BulkOperation -> L.ByteString
176−collapseStream stream = collapsed where
179+encodeBulkOperations stream = collapsed where
177− blobs = intersperse "\n" $ concat $ fmap getStreamChunk stream
180+ blobs = fmap encodeBulkOperation stream
178 mashedTaters = mash (mempty :: Builder) blobs
181 mashedTaters = mash (mempty :: Builder) blobs
179 collapsed = toLazyByteString $ mappend mashedTaters (byteString "\n")
182 collapsed = toLazyByteString $ mappend mashedTaters (byteString "\n")
180
183
181−mash :: Builder -> [L.ByteString] -> Builder
184+mash :: Builder -> V.Vector L.ByteString -> Builder
182−mash = foldl' (\b x -> mappend b (lazyByteString x))
185+mash = V.foldl' (\b x -> b `mappend` "\n" `mappend` (lazyByteString x))
183
186
184−mkMetadataValue :: Text -> String -> String -> String -> Value
187+mkBulkStreamValue :: Text -> String -> String -> String -> Value
185−mkMetadataValue operation indexName mappingName docId =
188+mkBulkStreamValue operation indexName mappingName docId =
186 object [operation .=
189 object [operation .=
187 object [ "_index" .= indexName
190 object [ "_index" .= indexName
188 , "_type" .= mappingName
191 , "_type" .= mappingName
189 , "_id" .= docId]]
192 , "_id" .= docId]]
190
193
191−getStreamChunk :: BulkOperation -> [L.ByteString]
194+encodeBulkOperation :: BulkOperation -> L.ByteString
192−getStreamChunk (BulkIndex (IndexName indexName)
195+encodeBulkOperation (BulkIndex (IndexName indexName)
193 (MappingName mappingName)
196 (MappingName mappingName)
194− (DocId docId) value) = blob where
197+ (DocId docId) value) = blob
195− metadata = mkMetadataValue "index" indexName mappingName docId
198+ where metadata = mkBulkStreamValue "index" indexName mappingName docId
196− blob = [encode metadata, encode value]
199+ blob = encode metadata `mappend` "\n" `mappend` encode value
197
200
198−getStreamChunk (BulkCreate (IndexName indexName)
201+encodeBulkOperation (BulkCreate (IndexName indexName)
199 (MappingName mappingName)
202 (MappingName mappingName)
200− (DocId docId) value) = blob where
203+ (DocId docId) value) = blob
201− metadata = mkMetadataValue "create" indexName mappingName docId
204+ where metadata = mkBulkStreamValue "create" indexName mappingName docId
202− blob = [encode metadata, encode value]
205+ blob = encode metadata `mappend` "\n" `mappend` encode value
203
206
204−getStreamChunk (BulkDelete (IndexName indexName)
207+encodeBulkOperation (BulkDelete (IndexName indexName)
205 (MappingName mappingName)
208 (MappingName mappingName)
206− (DocId docId)) = blob where
209+ (DocId docId)) = blob
207− metadata = mkMetadataValue "delete" indexName mappingName docId
210+ where metadata = mkBulkStreamValue "delete" indexName mappingName docId
208− blob = [encode metadata]
211+ blob = encode metadata
209
212
210−getStreamChunk (BulkUpdate (IndexName indexName)
213+encodeBulkOperation (BulkUpdate (IndexName indexName)
211 (MappingName mappingName)
214 (MappingName mappingName)
212− (DocId docId) value) = blob where
215+ (DocId docId) value) = blob
213− metadata = mkMetadataValue "update" indexName mappingName docId
216+ where metadata = mkBulkStreamValue "update" indexName mappingName docId
214− doc = object ["doc" .= value]
217+ doc = object ["doc" .= value]
215− blob = [encode metadata, encode doc]
218+ blob = encode metadata `mappend` "\n" `mappend` encode doc
219+
216
220
217getDocument :: Server -> IndexName -> MappingName
221getDocument :: Server -> IndexName -> MappingName
218 -> DocId -> IO Reply
222 -> DocId -> IO Reply
README.org+37−1
@@ -45,6 +45,7 @@ http://hackage.haskell.org/package/bloodhound
45-- the multi-line examples into your ghci REPL.
45-- the multi-line examples into your ghci REPL.
46
46
47:set -XDeriveGeneric
47:set -XDeriveGeneric
48+:{
48import Database.Bloodhound
49import Database.Bloodhound
49import Data.Aeson
50import Data.Aeson
50import Data.Either (Either(..))
51import Data.Either (Either(..))
@@ -68,6 +69,7 @@ let defaultIndexSettings = IndexSettings (ShardCount 3) (ReplicaCount 2)
68
69
69-- response :: Reply, Reply is a synonym for Network.HTTP.Conduit.Response
70-- response :: Reply, Reply is a synonym for Network.HTTP.Conduit.Response
70response <- createIndex testServer defaultIndexSettings testIndex
71response <- createIndex testServer defaultIndexSettings testIndex
72+:}
71
73
72#+END_SRC
74#+END_SRC
73
75
@@ -281,7 +283,7 @@ Right (Tweet {user = "bitemyapp"
281
283
282*** Bulk create, index
284*** Bulk create, index
283
285
284−#+BEGIN_SRC haskell
286+#+BEGIN_SRC haskell
285
287
286-- don't forget the imports and derive generic setting for ghci
288-- don't forget the imports and derive generic setting for ghci
287-- at the beginning of the examples.
289-- at the beginning of the examples.
@@ -318,6 +320,40 @@ resp <- bulk testServer stream
318
320
319#+END_SRC
321#+END_SRC
320
322
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+
321** Search
357** Search
322
358
323*** Querying
359*** Querying
tests/tests.hs+1−1
@@ -240,7 +240,7 @@ main = hspec $ do
240 testMapping (DocId "2") (toJSON firstTest)
240 testMapping (DocId "2") (toJSON firstTest)
241 let secondDoc = BulkCreate testIndex
241 let secondDoc = BulkCreate testIndex
242 testMapping (DocId "3") (toJSON secondTest)
242 testMapping (DocId "3") (toJSON secondTest)
243− let stream = [firstDoc, secondDoc]
243+ let stream = V.fromList [firstDoc, secondDoc]
244 _ <- bulk testServer stream
244 _ <- bulk testServer stream
245 _ <- refreshIndex testServer testIndex
245 _ <- refreshIndex testServer testIndex
246 fDoc <- getDocument testServer testIndex testMapping (DocId "2")
246 fDoc <- getDocument testServer testIndex testMapping (DocId "2")