UpdateClicksV2.groovy

package app

import io.elasticsearch.DocumentIndex
import io.mongodb.DocumentCollection
import org.bson.Document
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest
import org.elasticsearch.action.admin.indices.alias.get.GetAliasesResponse
import org.elasticsearch.client.Client
import org.elasticsearch.cluster.metadata.AliasMetaData
import org.elasticsearch.common.collect.ImmutableOpenMap
import org.elasticsearch.index.query.QueryBuilder
import org.elasticsearch.index.query.QueryBuilders
import org.elasticsearch.index.reindex.UpdateByQueryAction
import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder
import org.elasticsearch.script.Script
import org.elasticsearch.script.ScriptService
import util.PayoutFailsafe

/**
 * Created by dangnh@eway.vn on 11/3/17.
 */
class UpdateClicksV2 {
    static void main(String[] args) {

        def inhousePublishers = getPublisherInhouse()

        DocumentIndex clickIndex = new DocumentIndex('es://localhost:19300/clicks')

        def clickIndices = getIndicesFromAlias('clicks', clickIndex.delegate).sort().toArray(new String[0])


        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                                                 .must(QueryBuilders.termsQuery('publisher_id', inhousePublishers))
        def script = """ctx._source.inhouse = true;"""
        def result = PayoutFailsafe.instance.get {
            UpdateByQueryRequestBuilder updateRequestBuilder = UpdateByQueryAction.INSTANCE.newRequestBuilder(clickIndex.delegate)
            updateRequestBuilder.source(clickIndices)
                                .filter(queryBuilder)
                                .script(new Script(script))
                                .get()
        }
        if (result.indexingFailures){
            result.indexingFailures.each {println it.toString()}
        }

        if (result.searchFailures){
            result.searchFailures.each {println it.toString()}
        }
    }

    static List<String> getIndicesFromAlias(String alias, Client client) {
        ImmutableOpenMap<String, List<AliasMetaData>> aliases = ((GetAliasesResponse) client.admin().indices().getAliases(new GetAliasesRequest(alias)).actionGet()).getAliases()
        ArrayList<String> allIndices = new ArrayList()
        aliases.keysIt().forEachRemaining { index -> allIndices.add(index) }
        return allIndices
    }

    static Set<String> getOfferGuarantee() {
        DocumentCollection offerCollection = new DocumentCollection('mongodb://adflexmeta_ro:a45leXmRtarW@localhost:57017/adflex_meta.offers')

        def filter = [guarantee: true] as Document
        def projection = [id: 1] as Document

        offerCollection.find(filter).projection(projection).intoDocuments().thenApply { offers ->
            return (offers*.id as List<String>).toSet()
        }.join()
    }

    static Set<String> getPublisherInhouse() {
        DocumentCollection publisherCollection = new DocumentCollection('mongodb://adflexmeta_ro:a45leXmRtarW@localhost:57017/adflex_meta.publishers')

        def filter = [inhouse: true] as Document
        def projection = [id: 1] as Document
        publisherCollection.find(filter).projection(projection).intoDocuments().thenApply { publishers ->
            return (publishers*.id as List<String>).toSet()
        }.join()
    }
}