kafka-neo4j/src/main/java/com/winhc/service/impl/EsQueryServiceImpl.java

package com.winhc.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.winhc.pojo.MergePerson;
import com.winhc.service.EsQueryService;
import com.winhc.utils.EsQueryDsl;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.frameworkset.elasticsearch.boot.BBossESStarter;
import org.frameworkset.elasticsearch.client.ClientInterface;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.List;

/**
 * @author π
 * @Description:
 * @date 2022/5/25 19:22
 */
@Slf4j
@Service
@AllArgsConstructor
public class EsQueryServiceImpl implements EsQueryService {

    @Autowired
    @Qualifier("bbossESStarterEs5")
    private BBossESStarter bbossESStarterEs5;
    private ClientInterface restClient;

    @PostConstruct
    public void init() {
        restClient = bbossESStarterEs5.getRestClient("es5");
    }

    @Override
    public Integer queryByDsl(String human_pid) {
        String res = restClient.executeHttp("winhc_company_human_pid_mapping_v9/_count", EsQueryDsl.queryBoss(human_pid), ClientInterface.HTTP_POST);
        return JSONObject.parseObject(res).getInteger("count");
    }

    @Override
    public Collection<List<MergePerson>> queryByDsl(Collection<List<MergePerson>> mergePersonList) {
        mergePersonList.forEach(p -> p.forEach(m -> {
            m.setCnt(queryByDsl(m.getOld_human_pid()));
        }));
        return mergePersonList;
    }


}