@@ -42,6 +42,8 @@ module Database.Bloodhound.Client
42 , searchByIndex
42 , searchByIndex
43 , searchByType
43 , searchByType
44 , scanSearch
44 , scanSearch
45 , refreshIndex
47 , refreshIndex
46 , mkSearch
48 , mkSearch
47 , mkAggregateSearch
49 , mkAggregateSearch
@@ -78,6 +80,7 @@ import Data.Monoid
78import Data.Text (Text)
80import Data.Text (Text)
79import qualified Data.Text as T
81import qualified Data.Text as T
80import qualified Data.Text.Encoding as T
82import qualified Data.Text.Encoding as T
83+import Data.Time.Clock
81import qualified Data.Vector as V
84import qualified Data.Vector as V
82import Network.HTTP.Client
85import Network.HTTP.Client
83import qualified Network.HTTP.Types.Method as NHTM
86import qualified Network.HTTP.Types.Method as NHTM
@@ -608,8 +611,12 @@ searchByType (IndexName indexName)
608 (MappingName mappingName) = bindM2 dispatchSearch url . return
611 (MappingName mappingName) = bindM2 dispatchSearch url . return
609 where url = joinPath [indexName, mappingName, "_search"]
612 where url = joinPath [indexName, mappingName, "_search"]
614+-- | For a given scearch, request a scroll for efficient streaming of
615+-- search results. Note that the search is put into 'SearchTypeScan'
616+-- mode and thus results will not be sorted. Combine this with
617+-- 'advanceScroll' to efficiently stream through the full result set
611−scanSearch' :: MonadBH m => IndexName -> MappingName -> Search -> m (Maybe ScrollId)
618+getInitialScroll :: MonadBH m => IndexName -> MappingName -> Search -> m (Maybe ScrollId)
612−scanSearch' (IndexName indexName) (MappingName mappingName) search = do
619+getInitialScroll (IndexName indexName) (MappingName mappingName) search = do
613 let url = joinPath [indexName, mappingName, "_search"]
620 let url = joinPath [indexName, mappingName, "_search"]
614 search' = search { searchType = SearchTypeScan }
621 search' = search { searchType = SearchTypeScan }
615 resp' <- bindM2 dispatchSearch url (return search')
622 resp' <- bindM2 dispatchSearch url (return search')
@@ -617,31 +624,54 @@ scanSearch' (IndexName indexName) (MappingName mappingName) search = do
617 msid = maybe Nothing scrollId msr
624 msid = maybe Nothing scrollId msr
618 return msid
625 return msid
620−scroll' :: (FromJSON a, MonadBH m) => Maybe ScrollId -> m ([Hit a], Maybe ScrollId)
627+scroll' :: (FromJSON a, MonadBH m, MonadThrow m) => Maybe ScrollId -> m ([Hit a], Maybe ScrollId)
621scroll' Nothing = return ([], Nothing)
628scroll' Nothing = return ([], Nothing)
622scroll' (Just sid) = do
629scroll' (Just sid) = do
630+ res <- advanceScroll sid 60
632+ Right SearchResult {..} -> return (hits searchHits, scrollId)
633+ Left _ -> return ([], Nothing)
635+-- | Use the given scroll to fetch the next page of documents. If
636+-- there are still further pages, there will be a value in the
637+-- 'scrollId' field of the 'SearchResult'
645+ -- ^ How long should the snapshot of data be kept around? This timeout is updated every time 'advanceScroll' is used, so don't feel the need to set it to the entire duration of your search processing. Note that durations < 1s will be rounded up. Also note that 'NominalDiffTime' is an instance of Num so literals like 60 will be interpreted as seconds. 60s is a reasonable default.
646+ -> m (Either EsError (SearchResult a))
647+advanceScroll (ScrollId sid) scroll = do
623− url <- joinPath ["_search/scroll?scroll=1m"]
648+ url <- joinPath ["_search/scroll?scroll=" <> scrollTime]
624− resp' <- post url (Just . L.fromStrict $ T.encodeUtf8 sid)
649+ parseEsResponse =<< post url (Just . L.fromStrict $ T.encodeUtf8 sid)
625− let msr = decode' $ responseBody resp' :: FromJSON a => Maybe (SearchResult a)
650+ where scrollTime = showText secs <> "s"
626− resp = case msr of
651+ secs :: Integer
627− Just sr -> (hits $ searchHits sr, scrollId sr)
652+ secs = round scroll
628− _ -> ([], Nothing)
653+
631−simpleAccumilator :: (FromJSON a, MonadBH m) => [Hit a] -> ([Hit a], Maybe ScrollId) -> m ([Hit a], Maybe ScrollId)
654+simpleAccumulator :: (FromJSON a, MonadBH m, MonadThrow m) => [Hit a] -> ([Hit a], Maybe ScrollId) -> m ([Hit a], Maybe ScrollId)
632−simpleAccumilator oldHits (newHits, Nothing) = return (oldHits ++ newHits, Nothing)
655+simpleAccumulator oldHits (newHits, Nothing) = return (oldHits ++ newHits, Nothing)
633−simpleAccumilator oldHits ([], _) = return (oldHits, Nothing)
656+simpleAccumulator oldHits ([], _) = return (oldHits, Nothing)
634−simpleAccumilator oldHits (newHits, msid) = do
657+simpleAccumulator oldHits (newHits, msid) = do
635 (newHits', msid') <- scroll' msid
658 (newHits', msid') <- scroll' msid
636− simpleAccumilator (oldHits ++ newHits) (newHits', msid')
659+ simpleAccumulator (oldHits ++ newHits) (newHits', msid')
638-- | 'scanSearch' uses the 'scan&scroll' API of elastic,
661-- | 'scanSearch' uses the 'scan&scroll' API of elastic,
639−-- for a given 'IndexName' and 'MappingName',
662+-- for a given 'IndexName' and 'MappingName'. Note that this will
663+-- consume the entire search result set and will be doing O(n) list
664+-- appends so this may not be suitable for large result sets. In that
665+-- case, 'getInitialScroll' and 'advanceScroll' are good low level
666+-- tools. You should be able to hook them up trivially to conduit,
667+-- pipes, or your favorite streaming IO abstraction of choice. Note
668+-- that ordering on the search would destroy performance and thus is
640−scanSearch :: (FromJSON a, MonadBH m) => IndexName -> MappingName -> Search -> m [Hit a]
670+scanSearch :: (FromJSON a, MonadBH m, MonadThrow m) => IndexName -> MappingName -> Search -> m [Hit a]
641scanSearch indexName mappingName search = do
671scanSearch indexName mappingName search = do
642− msid <- scanSearch' indexName mappingName search
672+ msid <- getInitialScroll indexName mappingName search
643 (hits, msid') <- scroll' msid
673 (hits, msid') <- scroll' msid
644− (totalHits, _) <- simpleAccumilator [] (hits, msid')
674+ (totalHits, _) <- simpleAccumulator [] (hits, msid')
645 return totalHits
675 return totalHits
647-- | 'mkSearch' is a helper function for defaulting additional fields of a 'Search'
677-- | 'mkSearch' is a helper function for defaulting additional fields of a 'Search'