etl-camel-huge/src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelInsertA2Configuration.java

package com.gct.tools.etlcamelhuge.routeconfig;

import com.alibaba.fastjson.JSON;
import org.apache.camel.Message;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.model.ExpressionNode;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.RouteDefinition;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * class name: CamelJDBCConfiguration
 *
 * @author lloyd
 * @version 1.0
 * @since 2021/4/14 下午3:16
 */
@Configuration
public class CamelInsertA2Configuration {

    @Value("${insertA2.cron}")
    private String cron;


    @Bean
    public RouteBuilder routeBuilderWithInsertA2() {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("quartz:tab?cron="+cron)
                        .routeId("update-test")
                        .process(exchange -> {
                            Message in = exchange.getIn();
                            in.setHeader("date",getDate());
                        })
                        .setBody(simple("select well_id,prod_date,liq_prod_daily from aoid.aoid_daily_yield where prod_date='${header.date}' "))
                        .to("jdbc:aoid")
                        .split(body()).process(exchange -> {
                            Message in = exchange.getIn();
                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
                            //aRow.putIfAbsent("liq_prod_daily", null);
                            Timestamp timestamp = (Timestamp) aRow.get("prod_date");
                            Instant instant = Instant.ofEpochMilli(timestamp.getTime());
                            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
                            aRow.put("prod_date",localDateTime.plusDays(1).toLocalDate().format(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
                        })
                        .setBody(simple(" insert into DMS_DBA01_RCYL (JH, RQ, RCYL) values('${body[well_id]}', to_date(to_char('${body[prod_date]}'),'yyyy-MM-dd'),'${body[liq_prod_daily]}') "))
                        .to("jdbc:insertA2")
                        .end();
            };
        };
    }
    private String getDate(){
        return  LocalDateTime.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
    }
}