1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.technical.elasticsearch
21 
22 import akka.NotUsed
23 import akka.stream.scaladsl.Flow
24 import com.sksamuel.elastic4s.Index
25 import grizzled.slf4j.Logging
26 import org.make.api.post.PostSearchEngineComponent
27 import org.make.core.elasticsearch.IndexationStatus
28 import org.make.core.post.Post
29 import org.make.core.post.indexed.IndexedPost
30 
31 import scala.concurrent.Future
32 
33 trait PostIndexationStream extends IndexationStream with PostSearchEngineComponent with Logging {
34 
35   object PostStream {
36 
37     def flowIndexPosts(postIndexName: String): Flow[Post, IndexationStatus, NotUsed] =
38       grouped[Post].via(runIndexPost(postIndexName))
39 
40     def runIndexPost(postIndexName: String): Flow[Seq[Post], IndexationStatus, NotUsed] = {
41       Flow[Seq[Post]].mapAsync(singleAsync)(posts => executeIndexPosts(posts, postIndexName))
42     }
43 
44     private def executeIndexPosts(posts: Seq[Post], postIndexName: String): Future[IndexationStatus] = {
45       elasticsearchPostAPI
46         .indexPosts(posts.map(post => IndexedPost.createFromPost(post)), Some(Index(postIndexName)))
47     }
48 
49   }
50 }