package com.telerobot.fs.global; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.telerobot.fs.config.SystemConfig; import com.telerobot.fs.entity.po.CdrDetail; import com.telerobot.fs.entity.po.CdrEntity; import com.telerobot.fs.mybatis.dao.SysDao; import com.telerobot.fs.utils.CommonUtils; import com.telerobot.fs.utils.DateUtils; import com.telerobot.fs.utils.OkHttpClientUtil; import com.telerobot.fs.utils.StringUtils; import lombok.SneakyThrows; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.DependsOn; import org.springframework.stereotype.Component; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.Semaphore; @Component @DependsOn("appContextProvider") public class CdrPush implements ApplicationListener { @Autowired SysDao sysDao; private static Logger logger = LoggerFactory.getLogger(CdrPush.class); private static Semaphore semaphore = new Semaphore(9999); private static ArrayBlockingQueue cdrQueue = new ArrayBlockingQueue<>(9999); private static boolean checkPostCdrEnabled(){ return Boolean.parseBoolean(SystemConfig.getValue("post_cdr_enabled", "false")); } public static boolean addCdrToQueue(CdrDetail cdr){ if(!checkPostCdrEnabled()){ logger.info("{} cdr push is not enabled.", cdr.getUuid()); return false; } if(cdrQueue.add(cdr)){ semaphore.release(1); return true; }else{ logger.error("{} cdr-push queue is full. Cant not process new requests. cdr json={}", cdr.getUuid(), JSON.toJSONString(cdr) ); } return false; } private boolean postCdr(CdrDetail cdr){ try { String url = SystemConfig.getValue("post_cdr_url"); //如果有自定义回调地址 替换回调地址为自定义回调地址 否则默认回调 String callBackUrl = getCallBackUrlByUuid(cdr.getUuid()); logger.info("测试日志cdrUUid:{} ,callBackUrlByUuid: {}", cdr.getUuid(), callBackUrl); if(org.apache.commons.lang3.StringUtils.isNotBlank(callBackUrl)){ url = callBackUrl; } logger.info("测试日志最后请求的url:{}",url); if (StringUtils.isNullOrEmpty(url)) { logger.error("post_cdr_url has not been configured yet."); return false; } String cdrData = JSON.toJSONString(cdr); String response = OkHttpClientUtil.postCdr(url, cdrData); logger.info("{} postCdr: {}, request url {} , response: {}", cdr.getUuid(), cdrData, url, response); if ("success".equalsIgnoreCase(response)) { logger.info("{} post cdr succeed.", cdr.getUuid()); return true; } else { logger.error("{} post cdr failed: cdr data={}", cdr.getUuid(), cdrData); } }catch (Throwable e){ logger.error("postCdr failed: {} {}", e.toString(), CommonUtils.getStackTraceString(e.getStackTrace())); } return false; } @Override public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { new Thread(new Runnable() { @SneakyThrows @Override public void run() { try { logger.info("CdrPush thread is now running..."); while (true) { semaphore.acquire(); CdrDetail cdrDetail = cdrQueue.poll(); if (null != cdrDetail) { if(!postCdr(cdrDetail)){ addCdrToQueue(cdrDetail); } } Thread.sleep(10); } }catch (Throwable e){ logger.error("postCdr main thread error: {} {}", e.toString(), CommonUtils.getStackTraceString(e.getStackTrace())); } } }).start(); } private String getCallBackUrlByUuid(String uuid){ String userBizJson = sysDao.getUserBizJson(uuid); logger.info("测试日志userBizJson:{},查询uuid:{}", userBizJson,uuid); if(org.apache.commons.lang3.StringUtils.isNotBlank(userBizJson)){ JSONObject obj = JSONObject.parseObject(userBizJson); if(null != obj && !obj.isEmpty() && obj.containsKey("callbackUrl")){ return obj.getString("callbackUrl"); } } return null; } }