git.iko.soy/typeable/bloodhound
on master help sign in
bulk ops sans existential nonsense
6b50fdcChris Allen
Database/Bloodhound/Client.hs+50−19
compare with master
unifiedsplit
@@ -1,4 +1,4 @@
1−{-# LANGUAGE DeriveGeneric, ExistentialQuantification #-}
1+{-# LANGUAGE DeriveGeneric, ExistentialQuantification, StandaloneDeriving #-}
2
2
3module Database.Bloodhound.Client
3module Database.Bloodhound.Client
4 ( createIndex
4 ( createIndex
@@ -62,6 +62,7 @@ import Control.Monad (liftM)
62import Data.Aeson
62import Data.Aeson
63import Data.Aeson.TH (deriveJSON)
63import Data.Aeson.TH (deriveJSON)
64import qualified Data.ByteString.Lazy.Char8 as L
64import qualified Data.ByteString.Lazy.Char8 as L
65+import Data.ByteString.Builder
65import Data.List (intercalate)
66import Data.List (intercalate)
66import Data.Maybe (fromMaybe)
67import Data.Maybe (fromMaybe)
67import Data.Monoid
68import Data.Monoid
@@ -262,24 +263,54 @@ deleteDocument (Server server) (IndexName indexName)
262 method = NHTM.methodDelete
263 method = NHTM.methodDelete
263 body = Nothing
264 body = Nothing
264
265
265−-- bulk :: Server -> [BulkOperation a] -> IO Reply
266+bulk :: Server -> [BulkOperation] -> IO Reply
266−-- bulk (Server server) bulkOps = dispatch url method body where
267+bulk (Server server) bulkOps = dispatch url method body where
267−-- url = joinPath [server, "_bulk"]
268+ url = joinPath [server, "_bulk"]
268−-- method = NHTM.methodPost
269+ method = NHTM.methodPost
269−-- body = Just (L.intercalate "\n" . fmap encode $ bulkOps)
270+ blobs = concat $ fmap getStreamChunk bulkOps
270−
271+ body = Just $ toLazyByteString $ mash (mempty :: Builder) blobs
271−data BulkOperation a =
272+
272− forall a. ToJSON a =>
273+data BulkOperation =
273− BulkIndex (AssociatedDocument a)
274+ BulkIndex IndexName MappingName DocId Value
274− | BulkCreate (AssociatedDocument a)
275+ | BulkCreate IndexName MappingName DocId Value
275− | BulkDelete Association
276+ | BulkDelete IndexName MappingName DocId
276− | BulkUpdate (AssociatedDocument a)
277+ | BulkUpdate IndexName MappingName DocId Value deriving (Eq, Show)
277−
278+
278−data Placement = Placement IndexName MappingName deriving (Eq, Show)
279+mash :: Builder -> [L.ByteString] -> Builder
279−data Association = Association Placement DocId deriving (Eq, Show)
280+mash builder xs = foldr (\x b -> mappend b (lazyByteString x)) builder xs
280−
281+
281−data AssociatedDocument a =
282+mkMetadataValue :: Text -> String -> String -> String -> Value
282− forall a. ToJSON a => Associate Association a
283+mkMetadataValue operation indexName mappingName docId =
284+ object [operation .=
285+ object ["_index" .= indexName
286+ , "_type" .= mappingName
287+ , "_id" .= docId]]
288+
289+getStreamChunk :: BulkOperation -> [L.ByteString]
290+getStreamChunk (BulkIndex (IndexName indexName)
291+ (MappingName mappingName)
292+ (DocId docId) value) = blob where
293+ metadata = mkMetadataValue "index" indexName mappingName docId
294+ blob = [encode metadata, encode value]
295+
296+getStreamChunk (BulkCreate (IndexName indexName)
297+ (MappingName mappingName)
298+ (DocId docId) value) = blob where
299+ metadata = mkMetadataValue "create" indexName mappingName docId
300+ blob = [encode metadata, encode value]
301+
302+getStreamChunk (BulkDelete (IndexName indexName)
303+ (MappingName mappingName)
304+ (DocId docId)) = blob where
305+ metadata = mkMetadataValue "delete" indexName mappingName docId
306+ blob = [encode metadata]
307+
308+getStreamChunk (BulkUpdate (IndexName indexName)
309+ (MappingName mappingName)
310+ (DocId docId) value) = blob where
311+ metadata = mkMetadataValue "update" indexName mappingName docId
312+ doc = object ["doc" .= value]
313+ blob = [encode metadata, encode doc]
283
314
284data EsResult a = EsResult { _index :: Text
315data EsResult a = EsResult { _index :: Text
285 , _type :: Text
316 , _type :: Text