@@ -1,4 +1,4 @@
1−{-# LANGUAGE DeriveGeneric, ExistentialQuantification #-}
1+{-# LANGUAGE DeriveGeneric, ExistentialQuantification, StandaloneDeriving #-}
22
33module Database.Bloodhound.Client
44 ( createIndex
@@ -62,6 +62,7 @@ import Control.Monad (liftM)
6262import Data.Aeson
6363import Data.Aeson.TH (deriveJSON)
6464import qualified Data.ByteString.Lazy.Char8 as L
65+import Data.ByteString.Builder
6566import Data.List (intercalate)
6667import Data.Maybe (fromMaybe)
6768import Data.Monoid
@@ -262,24 +263,54 @@ deleteDocument (Server server) (IndexName indexName)
262263 method = NHTM.methodDelete
263264 body = Nothing
264265
265−-- bulk :: Server -> [BulkOperation a] -> IO Reply
266−-- bulk (Server server) bulkOps = dispatch url method body where
267−-- url = joinPath [server, "_bulk"]
268−-- method = NHTM.methodPost
269−-- body = Just (L.intercalate "\n" . fmap encode $ bulkOps)
270−
271−data BulkOperation a =
272− forall a. ToJSON a =>
273− BulkIndex (AssociatedDocument a)
274− | BulkCreate (AssociatedDocument a)
275− | BulkDelete Association
276− | BulkUpdate (AssociatedDocument a)
277−
278−data Placement = Placement IndexName MappingName deriving (Eq, Show)
279−data Association = Association Placement DocId deriving (Eq, Show)
280−
281−data AssociatedDocument a =
282− forall a. ToJSON a => Associate Association a
266+bulk :: Server -> [BulkOperation] -> IO Reply
267+bulk (Server server) bulkOps = dispatch url method body where
268+ url = joinPath [server, "_bulk"]
269+ method = NHTM.methodPost
270+ blobs = concat $ fmap getStreamChunk bulkOps
271+ body = Just $ toLazyByteString $ mash (mempty :: Builder) blobs
272+
273+data BulkOperation =
274+ BulkIndex IndexName MappingName DocId Value
275+ | BulkCreate IndexName MappingName DocId Value
276+ | BulkDelete IndexName MappingName DocId
277+ | BulkUpdate IndexName MappingName DocId Value deriving (Eq, Show)
278+
279+mash :: Builder -> [L.ByteString] -> Builder
280+mash builder xs = foldr (\x b -> mappend b (lazyByteString x)) builder xs
281+
282+mkMetadataValue :: Text -> String -> String -> String -> Value
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]
283314
284315data EsResult a = EsResult { _index :: Text
285316 , _type :: Text