-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2250 from FoxComm/search-service/interpreter-draft
Add dsl interpreter
- Loading branch information
Showing
15 changed files
with
360 additions
and
119 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,85 +1,87 @@ | ||
| package foxcomm.agni | ||
|
|
||
| import scala.language.postfixOps | ||
| import cats.implicits._ | ||
| import com.sksamuel.elastic4s.ElasticDsl._ | ||
| import com.sksamuel.elastic4s._ | ||
| import foxcomm.agni.dsl.query._ | ||
| import foxcomm.agni.interpreter.es._ | ||
| import io.circe._ | ||
| import io.circe.jawn.parseByteBuffer | ||
| import monix.eval.{Coeval, Task} | ||
| import org.elasticsearch.action.search.{SearchAction, SearchRequestBuilder, SearchResponse} | ||
| import org.elasticsearch.client.Client | ||
| import org.elasticsearch.client.transport.TransportClient | ||
| import org.elasticsearch.common.settings.Settings | ||
| import scala.concurrent.{ExecutionContext, Future} | ||
| import org.elasticsearch.common.transport.InetSocketTransportAddress | ||
| import org.elasticsearch.search.SearchHit | ||
|
|
||
| class SearchService(private val client: ElasticClient) extends AnyVal { | ||
| @SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements")) | ||
| class SearchService private (client: Client, qi: ESQueryInterpreter) { | ||
| import SearchService.ExtractJsonObject | ||
|
|
||
| def searchFor(searchIndex: IndexAndTypes, | ||
| def searchFor(searchIndex: String, | ||
| searchType: String, | ||
| searchQuery: SearchPayload, | ||
| searchSize: Int, | ||
| searchFrom: Option[Int])(implicit ec: ExecutionContext): Future[SearchResult] = { | ||
| val withQuery = searchQuery match { | ||
| case SearchPayload.es(query, _) ⇒ (_: SearchDefinition) rawQuery Json.fromJsonObject(query).noSpaces | ||
| searchFrom: Option[Int]): Task[SearchResult] = { | ||
| def prepareBuilder: Coeval[SearchRequestBuilder] = Coeval.eval { | ||
| val builder = new SearchRequestBuilder(client, SearchAction.INSTANCE) | ||
| builder | ||
| .setIndices(searchIndex) | ||
| .setTypes(searchType) | ||
| .setSize(searchSize) | ||
| searchFrom.foreach(builder.setFrom) | ||
| searchQuery.fields.foreach(fs ⇒ builder.setFetchSource(fs.toList.toArray, Array.empty[String])) | ||
| builder | ||
| } | ||
|
|
||
| def evalQuery(builder: SearchRequestBuilder): Coeval[SearchRequestBuilder] = searchQuery match { | ||
| case SearchPayload.es(query, _) ⇒ | ||
| Coeval.eval(builder.setQuery(Json.fromJsonObject(query).dump)) | ||
| case SearchPayload.fc(query, _) ⇒ | ||
| // TODO: this is really some basic and quite ugly interpreter | ||
| // consider more principled approach | ||
| // maybe free monad would be a good fit there? | ||
| (_: SearchDefinition) bool { | ||
| query | ||
| .map(_.query.foldLeft(new BoolQueryDefinition) { | ||
| case (bool, QueryFunction.eq(in, value)) ⇒ | ||
| bool.filter(in.toList.map(termsQuery(_, value.toList: _*))) | ||
| case (bool, QueryFunction.neq(in, value)) ⇒ | ||
| bool.not(in.toList.map(termsQuery(_, value.toList: _*))) | ||
| case (bool, QueryFunction.matches(in, value)) ⇒ | ||
| val fields = in.toList | ||
| bool.must(value.toList.map(q ⇒ multiMatchQuery(q).fields(fields))) | ||
| case (bool, QueryFunction.range(in, value)) ⇒ | ||
| val query = rangeQuery(in.field) | ||
| val unified = value.unify | ||
| val queryWithLowerBound = unified.lower.fold(query) { | ||
| case (b, v) ⇒ query.from(v).includeLower(b.withBound) | ||
| } | ||
| val boundedQuery = unified.upper.fold(queryWithLowerBound) { | ||
| case (b, v) ⇒ queryWithLowerBound.to(v).includeUpper(b.withBound) | ||
| } | ||
| bool.filter(boundedQuery) | ||
| case (bool, _) ⇒ bool // TODO: implement rest of cases | ||
| }) | ||
| .getOrElse((new BoolQueryDefinition).must(matchAllQuery)) | ||
| } | ||
| qi(query).map(builder.setQuery) | ||
| } | ||
| val baseSearch = withQuery(search in searchIndex size searchSize) | ||
| val limitedSearch = | ||
| searchQuery.fields.fold(baseSearch)(fields ⇒ baseSearch sourceInclude (fields.toList: _*)) | ||
| client | ||
| .execute(searchFrom.fold(limitedSearch)(limitedSearch from)) | ||
| .map(response ⇒ | ||
| SearchResult( | ||
| result = response.hits.collect { | ||
|
|
||
| def setupBuilder: Task[SearchRequestBuilder] = (prepareBuilder flatMap evalQuery).task | ||
|
|
||
| for { | ||
| builder ← setupBuilder | ||
| request = builder.request() | ||
| response ← async[SearchResponse, SearchResult](client.search(request, _)) | ||
| } yield { | ||
| val hits = response.getHits | ||
| SearchResult( | ||
| result = hits | ||
| .hits() | ||
| .view | ||
| .collect { | ||
| case ExtractJsonObject(obj) ⇒ obj | ||
| }(collection.breakOut), | ||
| pagination = SearchPagination(total = response.totalHits), | ||
| maxScore = response.maxScore | ||
| )) | ||
| } | ||
| .toList, | ||
| pagination = SearchPagination(total = hits.totalHits()), | ||
| maxScore = hits.getMaxScore | ||
| ) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| object SearchService { | ||
| object ExtractJsonObject { | ||
| def unapply(hit: RichSearchHit): Option[JsonObject] = | ||
| def unapply(hit: SearchHit): Option[JsonObject] = | ||
| parseByteBuffer(hit.sourceRef.toChannelBuffer.toByteBuffer).toOption | ||
| .flatMap(_.asObject) | ||
| } | ||
|
|
||
| def apply(client: ElasticClient): SearchService = new SearchService(client) | ||
| def apply(client: Client, qi: ESQueryInterpreter): SearchService = | ||
| new SearchService(client, qi) | ||
|
|
||
| def fromConfig(config: AppConfig): SearchService = { | ||
| def fromConfig(config: AppConfig, qi: ESQueryInterpreter): SearchService = { | ||
| val esConfig = config.elasticsearch | ||
| val settings = | ||
| Settings.settingsBuilder().put("cluster.name", esConfig.cluster).build() | ||
| val client = | ||
| ElasticClient.transport(settings, ElasticsearchClientUri(esConfig.host)) | ||
| val client = TransportClient | ||
| .builder() | ||
| .settings(settings) | ||
| .build() | ||
| .addTransportAddresses(esConfig.host.toList.map(new InetSocketTransportAddress(_)): _*) | ||
|
|
||
| new SearchService(client) | ||
| apply(client, qi) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.