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

package com.winhc.service.impl;

import com.winhc.service.RelationService;
import com.winhc.utils.CompanyUtils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.neo4j.driver.Driver;
import org.neo4j.driver.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author π
 * @Description:案件-更新节点
 * @date 2021/5/17 10:03
 */
@Slf4j
@Service("caseNodeUpdateImpl")
@AllArgsConstructor
public class CaseNodeUpdateImpl implements RelationService {

    @Autowired
    @Qualifier("DriverV2")
    Driver driver;

    @Override
    public String save(List<Map<String, Object>> batch_list) {
        if (batch_list.isEmpty()) return null;
        long start = System.currentTimeMillis();
        Session session = driver.session();
//        final String cql = "\nWITH  {batch_list} AS batch_list \n" +
//                "UNWIND batch_list AS row \n" +
//                " MERGE(s:CASE{case_id:row.start_id})\n" +
//                " SET s.case_id = row.start_id\n" +
//                " WITH s\n" +
//                " CALL apoc.path.subgraphNodes(s, {maxLevel:-1}) YIELD node\n" +
//                " WITH node,s\n" +
//                " SET node.component_id = -id(s)\n" +
//                " SET node:" + CompanyUtils.getIncrPersonLabelV2("新增");

        final String cql = "\nWITH  {batch_list} AS batch_list \n" +
                "UNWIND batch_list AS row\n" +
                "MERGE (s:CASE{case_id:row.start_id})\n" +
                "WITH s\n" +
                "CALL apoc.path.subgraphNodes(s, {maxLevel:-1}) YIELD node\n" +
                "WITH ID(s) as id,apoc.coll.max(collect(node.component_id) + collect(-ID(s))) as update_id,collect(node) as nodes\n" +
                "UNWIND nodes AS n\n" +
                "SET n.component_id = update_id\n" +
                "SET n:" + CompanyUtils.getIncrPersonLabelV2("新增");
        log.info("consumer size: {}, cql:{}", batch_list.size(), cql);
        String data = CompanyUtils.writeNeo4j(session, cql, new HashMap<String, Object>() {{
            put("batch_list", batch_list);
        }});
        session.close();
        log.info("class:{} | save size:{} | cost:{}", CaseNodeUpdateImpl.class.getSimpleName(), batch_list.size(), (System.currentTimeMillis() - start));
        return data;
    }
}