Merge branch 'feature/inflow-control-improvement' into master
This commit is contained in:
curry772
@@ -41,20 +41,36 @@ public class InflowControlDAO extends BaseDAO {
|
||||
@Autowired
|
||||
private InflowControlGroupMappingLoader inflowControlGroupMappingLoader;
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Enum 기반 범용 메서드 — type 코드를 직접 노출하지 않음
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
public List<InflowTargetVO> getInflowList(InflowType type) throws DAOException {
|
||||
return getInflowList(type.code());
|
||||
}
|
||||
|
||||
public Map<String, InflowTargetVO> getInflowMap(InflowType type, String name) throws DAOException {
|
||||
return getInflowMap(type.code(), name);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 기존 명명 메서드 — enum 기반 메서드로 위임
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
public List<InflowTargetVO> getInflowAdapterList() throws DAOException {
|
||||
return getInflowList("01");
|
||||
return getInflowList(InflowType.ADAPTER);
|
||||
}
|
||||
|
||||
public List<InflowTargetVO> getInflowInterfaceList() throws DAOException {
|
||||
return getInflowList("02");
|
||||
return getInflowList(InflowType.INTERFACE);
|
||||
}
|
||||
|
||||
public Map<String, InflowTargetVO> getInflowTargetByAdater(String name) throws DAOException {
|
||||
return getInflowMap("01", name);
|
||||
return getInflowMap(InflowType.ADAPTER, name);
|
||||
}
|
||||
|
||||
public Map<String, InflowTargetVO> getInflowTargetByInterface(String name) throws DAOException {
|
||||
return getInflowMap("02", name);
|
||||
return getInflowMap(InflowType.INTERFACE, name);
|
||||
}
|
||||
|
||||
private List<InflowTargetVO> getInflowList(String type) throws DAOException {
|
||||
|
||||
@@ -7,6 +7,7 @@ import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
@@ -39,11 +40,11 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
public static final String QUOTA_TIMEUNIT_DAY = "DAY";
|
||||
public static final String QUOTA_TIMEUNIT_MONTH = "MON";
|
||||
|
||||
private Map<String, CustomBucket> adapterBucketList = new HashMap<>();
|
||||
private Map<String, CustomBucket> interfaceBucketList = new HashMap<>();
|
||||
private Map<String, CustomGroupBucket> groupBucketList = new HashMap<>();
|
||||
private Map<String, String> interfaceToGroupMap = new HashMap<>();
|
||||
private Map<String, InflowGroupVO> groupVoMap = new HashMap<>();
|
||||
private volatile Map<String, CustomBucket> adapterBucketList = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, CustomBucket> interfaceBucketList = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, CustomGroupBucket> groupBucketList = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, String> interfaceToGroupMap = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, InflowGroupVO> groupVoMap = new ConcurrentHashMap<>();
|
||||
private boolean started;
|
||||
private LifecycleSupport lifecycle = new LifecycleSupport(this);
|
||||
|
||||
@@ -97,37 +98,39 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
}
|
||||
|
||||
private void initAdapter() throws Exception {
|
||||
adapterBucketList = new HashMap<>();
|
||||
Map<String, CustomBucket> newMap = new HashMap<>();
|
||||
List<InflowTargetVO> inflowAdapterVoList = inflowControlDAO.getInflowAdapterList();
|
||||
|
||||
for (InflowTargetVO inflowVo : inflowAdapterVoList) {
|
||||
if (InflowControlManager.isInflowTarget(inflowVo)) {
|
||||
CustomBucket bucket = makeBucketUsingInflowVO(inflowVo);
|
||||
if (bucket != null) {
|
||||
adapterBucketList.put(inflowVo.getName(), bucket);
|
||||
newMap.put(inflowVo.getName(), bucket);
|
||||
}
|
||||
}
|
||||
}
|
||||
this.adapterBucketList = newMap;
|
||||
}
|
||||
|
||||
private void initInterface() throws Exception {
|
||||
interfaceBucketList = new HashMap<>();
|
||||
Map<String, CustomBucket> newMap = new HashMap<>();
|
||||
List<InflowTargetVO> inflowInterfaceVoList = inflowControlDAO.getInflowInterfaceList();
|
||||
|
||||
for (InflowTargetVO inflowVo : inflowInterfaceVoList) {
|
||||
if (InflowControlManager.isInflowTarget(inflowVo)) {
|
||||
CustomBucket bucket = makeBucketUsingInflowVO(inflowVo);
|
||||
if (bucket != null) {
|
||||
interfaceBucketList.put(inflowVo.getName(), bucket);
|
||||
newMap.put(inflowVo.getName(), bucket);
|
||||
}
|
||||
}
|
||||
}
|
||||
this.interfaceBucketList = newMap;
|
||||
}
|
||||
|
||||
private void initGroup() throws Exception {
|
||||
groupBucketList = new HashMap<>();
|
||||
interfaceToGroupMap = new HashMap<>();
|
||||
groupVoMap = new HashMap<>();
|
||||
Map<String, CustomGroupBucket> newGroupBucketList = new HashMap<>();
|
||||
Map<String, String> newInterfaceToGroupMap = new HashMap<>();
|
||||
Map<String, InflowGroupVO> newGroupVoMap = new HashMap<>();
|
||||
|
||||
List<InflowGroupVO> inflowGroupVoList = inflowControlDAO.getInflowGroupList();
|
||||
|
||||
@@ -135,20 +138,24 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
if (isInflowGroupTarget(groupVo)) {
|
||||
CustomGroupBucket bucket = makeGroupBucket(groupVo);
|
||||
if (bucket != null) {
|
||||
groupBucketList.put(groupVo.getGroupId(), bucket);
|
||||
groupVoMap.put(groupVo.getGroupId(), groupVo);
|
||||
newGroupBucketList.put(groupVo.getGroupId(), bucket);
|
||||
newGroupVoMap.put(groupVo.getGroupId(), groupVo);
|
||||
// 인터페이스 → 그룹 매핑 등록
|
||||
for (String interfaceId : groupVo.getInterfaceList()) {
|
||||
interfaceToGroupMap.put(interfaceId, groupVo.getGroupId());
|
||||
newInterfaceToGroupMap.put(interfaceId, groupVo.getGroupId());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (logger.isInfo()) {
|
||||
logger.info("InflowControlManager] initGroup completed. groups=" + groupBucketList.size()
|
||||
+ ", mappings=" + interfaceToGroupMap.size());
|
||||
logger.info("InflowControlManager] initGroup completed. groups=" + newGroupBucketList.size()
|
||||
+ ", mappings=" + newInterfaceToGroupMap.size());
|
||||
}
|
||||
|
||||
this.groupBucketList = newGroupBucketList;
|
||||
this.interfaceToGroupMap = newInterfaceToGroupMap;
|
||||
this.groupVoMap = newGroupVoMap;
|
||||
}
|
||||
|
||||
public synchronized void reloadAdapter() throws Exception {
|
||||
@@ -293,11 +300,11 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
// Notify our interested LifecycleListeners
|
||||
lifecycle.fireLifecycleEvent(STOPING_EVENT, this);
|
||||
|
||||
adapterBucketList.clear();
|
||||
interfaceBucketList.clear();
|
||||
groupBucketList.clear();
|
||||
interfaceToGroupMap.clear();
|
||||
groupVoMap.clear();
|
||||
adapterBucketList = new ConcurrentHashMap<>();
|
||||
interfaceBucketList = new ConcurrentHashMap<>();
|
||||
groupBucketList = new ConcurrentHashMap<>();
|
||||
interfaceToGroupMap = new ConcurrentHashMap<>();
|
||||
groupVoMap = new ConcurrentHashMap<>();
|
||||
|
||||
started = false;
|
||||
// Notify our interested LifecycleListeners
|
||||
@@ -340,15 +347,15 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
return svcCd;
|
||||
}
|
||||
|
||||
public void removeAdapter(String adapter) {
|
||||
public synchronized void removeAdapter(String adapter) {
|
||||
adapterBucketList.remove(adapter);
|
||||
}
|
||||
|
||||
public void removeInterface(String inter) {
|
||||
public synchronized void removeInterface(String inter) {
|
||||
interfaceBucketList.remove(inter);
|
||||
}
|
||||
|
||||
public void removeGroup(String groupId) {
|
||||
public synchronized void removeGroup(String groupId) {
|
||||
groupBucketList.remove(groupId);
|
||||
groupVoMap.remove(groupId);
|
||||
// 해당 그룹에 속한 인터페이스 매핑도 제거
|
||||
@@ -371,6 +378,10 @@ public class InflowControlManager implements Lifecycle, Bucket {
|
||||
return groupIds;
|
||||
}
|
||||
|
||||
public CustomGroupBucket getGroupBucket(String groupId) {
|
||||
return groupBucketList.get(groupId);
|
||||
}
|
||||
|
||||
private CustomBucket makeBucketUsingInflowVO(InflowTargetVO inflowTarget) {
|
||||
if (InflowControlManager.isInflowTarget(inflowTarget)) {
|
||||
LocalBucketBuilder builder = Bucket4j.builder();
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
package com.eactive.eai.common.inflow;
|
||||
|
||||
/**
|
||||
* 유량제어 대상 유형.
|
||||
* {@link InflowControlDAO}의 type 컬럼 값을 enum으로 관리하여
|
||||
* 호출부에 코드 문자열이 노출되지 않도록 한다.
|
||||
*/
|
||||
public enum InflowType {
|
||||
|
||||
ADAPTER("01"),
|
||||
INTERFACE("02"),
|
||||
CLIENT("03");
|
||||
|
||||
private final String code;
|
||||
|
||||
InflowType(String code) {
|
||||
this.code = code;
|
||||
}
|
||||
|
||||
public String code() {
|
||||
return code;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import com.eactive.eai.common.inflow.Bucket;
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
|
||||
/**
|
||||
* Bucket 인터페이스 확장.
|
||||
* 기존 isAdapterPass/isInterfacePass(boolean)에 더해
|
||||
* 어느 버킷(PER_SECOND/THRESHOLD)에서 차단됐는지 반환하는 메서드 추가.
|
||||
* 클라이언트별 유량제어 메서드도 포함.
|
||||
*/
|
||||
public interface DualBucket extends Bucket {
|
||||
|
||||
/**
|
||||
* 어댑터 유량제어 체크 — 차단 원인 포함.
|
||||
* @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
|
||||
*/
|
||||
String isAdapterPassDetail(String adapter);
|
||||
|
||||
/**
|
||||
* 인터페이스 유량제어 체크 — 차단 원인 포함.
|
||||
* @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
|
||||
*/
|
||||
String isInterfacePassDetail(String inter);
|
||||
|
||||
/**
|
||||
* 클라이언트 유량제어 체크 — 차단 원인 포함.
|
||||
* @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
|
||||
*/
|
||||
String isClientPassDetail(String clientId);
|
||||
|
||||
/**
|
||||
* 클라이언트 유량제어 설정 조회 — 에러 메시지 생성용.
|
||||
* @return 등록된 설정이 없으면 null
|
||||
*/
|
||||
InflowTargetVO getClientInflowThreshold(String clientId);
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
/**
|
||||
* 어댑터/인터페이스 유량제어 버킷.
|
||||
* 기존 CustomBucket(단일 버킷)과 달리 perSecond/threshold를 별도 버킷으로 분리하여
|
||||
* 어느 한도를 초과했는지 구분 가능 (CustomGroupBucket과 동일한 이중 구조).
|
||||
*/
|
||||
public class DualCustomBucket {
|
||||
|
||||
public static final String RESULT_PASS = null;
|
||||
public static final String RESULT_BLOCKED_PER_SECOND = "PER_SECOND";
|
||||
public static final String RESULT_BLOCKED_THRESHOLD = "THRESHOLD";
|
||||
|
||||
private final LocalBucket perSecondBucket;
|
||||
private final LocalBucket thresholdBucket;
|
||||
private final InflowTargetVO inflowTargetVo;
|
||||
|
||||
public DualCustomBucket(LocalBucket perSecondBucket, LocalBucket thresholdBucket, InflowTargetVO inflowTargetVo) {
|
||||
this.perSecondBucket = perSecondBucket;
|
||||
this.thresholdBucket = thresholdBucket;
|
||||
this.inflowTargetVo = inflowTargetVo;
|
||||
}
|
||||
|
||||
/**
|
||||
* 두 버킷을 순서대로 소비 시도.
|
||||
* @return null이면 통과, "PER_SECOND" 또는 "THRESHOLD"이면 해당 버킷에서 차단
|
||||
*/
|
||||
public String tryConsume() {
|
||||
boolean perSecondPass = (perSecondBucket == null) || perSecondBucket.tryConsume(1);
|
||||
if (!perSecondPass) {
|
||||
return RESULT_BLOCKED_PER_SECOND;
|
||||
}
|
||||
|
||||
boolean thresholdPass = (thresholdBucket == null) || thresholdBucket.tryConsume(1);
|
||||
if (!thresholdPass) {
|
||||
// perSecond 토큰은 이미 소비됨 — Token Bucket 특성상 롤백 불가
|
||||
// 초당 버킷은 빠르게 리필되므로 실질적 영향 미미
|
||||
return RESULT_BLOCKED_THRESHOLD;
|
||||
}
|
||||
|
||||
return RESULT_PASS;
|
||||
}
|
||||
|
||||
public long getPerSecondAvailableTokens() {
|
||||
return (perSecondBucket != null) ? perSecondBucket.getAvailableTokens() : -1;
|
||||
}
|
||||
|
||||
public long getThresholdAvailableTokens() {
|
||||
return (thresholdBucket != null) ? thresholdBucket.getAvailableTokens() : -1;
|
||||
}
|
||||
|
||||
public LocalBucket getPerSecondBucket() { return perSecondBucket; }
|
||||
public LocalBucket getThresholdBucket() { return thresholdBucket; }
|
||||
public InflowTargetVO getInflowTargetVo() { return inflowTargetVo; }
|
||||
}
|
||||
@@ -0,0 +1,313 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import com.eactive.eai.common.inflow.Bucket;
|
||||
import com.eactive.eai.common.inflow.CustomGroupBucket;
|
||||
import com.eactive.eai.common.inflow.InflowControlDAO;
|
||||
import com.eactive.eai.common.inflow.InflowControlManager;
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
import com.eactive.eai.common.inflow.InflowType;
|
||||
import com.eactive.eai.common.lifecycle.LifecycleException;
|
||||
import com.eactive.eai.common.message.EAIMessage;
|
||||
import com.eactive.eai.common.server.EAIServerManager;
|
||||
import com.eactive.eai.common.util.ApplicationContextProvider;
|
||||
import com.eactive.eai.common.util.Logger;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket4j;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
/**
|
||||
* 어댑터/인터페이스/클라이언트 버킷을 이중 구조(perSecond + threshold)로 관리하는 유량제어 관리자.
|
||||
*
|
||||
* <p>기존 InflowControlManager는 어댑터/인터페이스에 단일 LocalBucket(복수 Bandwidth)을 사용하여
|
||||
* 차단 시 어느 한도(초당/기간)를 초과했는지 알 수 없었다.
|
||||
* 이 클래스는 그룹 버킷(CustomGroupBucket)과 동일하게 두 버킷을 분리하고
|
||||
* {@link DualBucket#isAdapterPassDetail}/{@link DualBucket#isInterfacePassDetail}로
|
||||
* 차단 원인을 반환한다. 그룹 버킷 통과 여부는 {@link TargetMetrics}로 카운팅된다.
|
||||
*
|
||||
* <p>적용 방법: INFLOW.properties에 아래 한 줄 추가.
|
||||
* <pre>inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager</pre>
|
||||
*/
|
||||
@Component
|
||||
public class DualInflowControlManager extends InflowControlManager implements DualBucket {
|
||||
|
||||
private static final Logger logger = Logger.getLogger(Logger.LOGGER_DEFAULT);
|
||||
|
||||
// InflowControlUtil이 리플렉션으로 호출하는 진입점
|
||||
public static synchronized Bucket getBucket() {
|
||||
return ApplicationContextProvider.getContext().getBean(DualInflowControlManager.class);
|
||||
}
|
||||
|
||||
public static Boolean isTargetOfInflowControl(EAIServerManager eaiServerManager, EAIMessage eaiMessage) {
|
||||
return InflowControlManager.isTargetOfInflowControl(eaiServerManager, eaiMessage);
|
||||
}
|
||||
|
||||
@Autowired
|
||||
private InflowControlDAO dualDao;
|
||||
|
||||
private volatile Map<String, DualCustomBucket> adapterBucketMap = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, DualCustomBucket> interfaceBucketMap = new ConcurrentHashMap<>();
|
||||
private volatile Map<String, DualCustomBucket> clientBucketMap = new ConcurrentHashMap<>();
|
||||
|
||||
private final ConcurrentHashMap<String, TargetMetrics> groupMetricsMap = new ConcurrentHashMap<>();
|
||||
|
||||
public static class TargetMetrics {
|
||||
public final AtomicLong allowed = new AtomicLong();
|
||||
public final AtomicLong rejectedPerSecond = new AtomicLong();
|
||||
public final AtomicLong rejectedThreshold = new AtomicLong();
|
||||
public volatile long lastResetTime = System.currentTimeMillis();
|
||||
|
||||
public void reset() {
|
||||
allowed.set(0);
|
||||
rejectedPerSecond.set(0);
|
||||
rejectedThreshold.set(0);
|
||||
lastResetTime = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Lifecycle
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Override
|
||||
public void start() throws LifecycleException {
|
||||
super.start();
|
||||
try {
|
||||
initAdapters();
|
||||
initInterfaces();
|
||||
initClients();
|
||||
} catch (Exception e) {
|
||||
throw new LifecycleException("DualInflowControlManager init failed: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 초기화 / 리로드
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private void initAdapters() throws Exception {
|
||||
this.adapterBucketMap = loadBucketMap(InflowType.ADAPTER);
|
||||
if (logger.isInfo()) logger.info("DualInflowControlManager] initAdapters: " + adapterBucketMap.size() + " buckets");
|
||||
}
|
||||
|
||||
private void initInterfaces() throws Exception {
|
||||
this.interfaceBucketMap = loadBucketMap(InflowType.INTERFACE);
|
||||
if (logger.isInfo()) logger.info("DualInflowControlManager] initInterfaces: " + interfaceBucketMap.size() + " buckets");
|
||||
}
|
||||
|
||||
private void initClients() throws Exception {
|
||||
this.clientBucketMap = loadBucketMap(InflowType.CLIENT);
|
||||
if (logger.isInfo()) logger.info("DualInflowControlManager] initClients: " + clientBucketMap.size() + " buckets");
|
||||
}
|
||||
|
||||
private Map<String, DualCustomBucket> loadBucketMap(InflowType type) throws Exception {
|
||||
Map<String, DualCustomBucket> newMap = new HashMap<>();
|
||||
for (InflowTargetVO vo : dualDao.getInflowList(type)) {
|
||||
DualCustomBucket bucket = makeBucket(vo);
|
||||
if (bucket != null) newMap.put(vo.getName(), bucket);
|
||||
}
|
||||
return newMap;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadAdapter() throws Exception {
|
||||
super.reloadAdapter();
|
||||
initAdapters();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadAdapter(String adapter) throws Exception {
|
||||
super.reloadAdapter(adapter);
|
||||
reloadBucketMap(adapterBucketMap, InflowType.ADAPTER, adapter);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadInterface() throws Exception {
|
||||
super.reloadInterface();
|
||||
initInterfaces();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadInterface(String inter) throws Exception {
|
||||
super.reloadInterface(inter);
|
||||
reloadBucketMap(interfaceBucketMap, InflowType.INTERFACE, inter);
|
||||
}
|
||||
|
||||
public synchronized void reloadClient() throws Exception {
|
||||
initClients();
|
||||
}
|
||||
|
||||
public synchronized void reloadClient(String clientId) throws Exception {
|
||||
reloadBucketMap(clientBucketMap, InflowType.CLIENT, clientId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadGroup() throws Exception {
|
||||
super.reloadGroup();
|
||||
groupMetricsMap.clear();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void reloadGroup(String groupId) throws Exception {
|
||||
super.reloadGroup(groupId);
|
||||
TargetMetrics m = groupMetricsMap.get(groupId);
|
||||
if (m != null) m.reset();
|
||||
}
|
||||
|
||||
private void reloadBucketMap(Map<String, DualCustomBucket> targetMap, InflowType type, String name) throws Exception {
|
||||
Map<String, InflowTargetVO> updated = dualDao.getInflowMap(type, name);
|
||||
if (updated == null) return;
|
||||
for (InflowTargetVO vo : updated.values()) {
|
||||
DualCustomBucket bucket = makeBucket(vo);
|
||||
if (bucket != null) targetMap.put(vo.getName(), bucket);
|
||||
else targetMap.remove(vo.getName());
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 그룹 메트릭 — isGroupPass() 카운팅 및 접근자
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Override
|
||||
public String isGroupPass(String groupId) {
|
||||
String result = super.isGroupPass(groupId);
|
||||
TargetMetrics m = groupMetricsMap.computeIfAbsent(groupId, k -> new TargetMetrics());
|
||||
if (result == null) {
|
||||
m.allowed.incrementAndGet();
|
||||
} else if (CustomGroupBucket.RESULT_BLOCKED_PER_SECOND.equals(result)) {
|
||||
m.rejectedPerSecond.incrementAndGet();
|
||||
} else {
|
||||
m.rejectedThreshold.incrementAndGet();
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public TargetMetrics getGroupMetrics(String groupId) {
|
||||
return groupMetricsMap.get(groupId);
|
||||
}
|
||||
|
||||
public Map<String, TargetMetrics> getAllGroupMetrics() {
|
||||
return Collections.unmodifiableMap(groupMetricsMap);
|
||||
}
|
||||
|
||||
public void resetGroupMetrics(String groupId) {
|
||||
TargetMetrics m = groupMetricsMap.get(groupId);
|
||||
if (m != null) m.reset();
|
||||
}
|
||||
|
||||
public void resetAllGroupMetrics() {
|
||||
groupMetricsMap.values().forEach(TargetMetrics::reset);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// DualBucket — 차단 원인 포함 메서드
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Override
|
||||
public String isAdapterPassDetail(String adapter) {
|
||||
DualCustomBucket b = adapterBucketMap.get(adapter);
|
||||
return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String isInterfacePassDetail(String inter) {
|
||||
DualCustomBucket b = interfaceBucketMap.get(inter);
|
||||
return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String isClientPassDetail(String clientId) {
|
||||
DualCustomBucket b = clientBucketMap.get(clientId);
|
||||
return (b == null) ? DualCustomBucket.RESULT_PASS : b.tryConsume();
|
||||
}
|
||||
|
||||
@Override
|
||||
public InflowTargetVO getClientInflowThreshold(String clientId) {
|
||||
DualCustomBucket b = clientBucketMap.get(clientId);
|
||||
return (b == null) ? null : b.getInflowTargetVo();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Bucket 인터페이스 오버라이드 — 이중 버킷으로 교체 (boolean 반환 유지)
|
||||
// isAdapterPassDetail/isInterfacePassDetail이 토큰을 소비하므로
|
||||
// 두 메서드를 동일 요청에서 중복 호출하면 토큰이 이중 소비됨에 주의.
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Override
|
||||
public boolean isAdapterPass(String adapter) {
|
||||
return isAdapterPassDetail(adapter) == DualCustomBucket.RESULT_PASS;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isInterfacePass(String inter) {
|
||||
return isInterfacePassDetail(inter) == DualCustomBucket.RESULT_PASS;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 모니터링용 접근자
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
public DualCustomBucket getAdapterBucket(String adapter) {
|
||||
return adapterBucketMap.get(adapter);
|
||||
}
|
||||
|
||||
public DualCustomBucket getInterfaceBucket(String inter) {
|
||||
return interfaceBucketMap.get(inter);
|
||||
}
|
||||
|
||||
public Map<String, DualCustomBucket> getAdapterBucketMap() {
|
||||
return adapterBucketMap;
|
||||
}
|
||||
|
||||
public Map<String, DualCustomBucket> getInterfaceBucketMap() {
|
||||
return interfaceBucketMap;
|
||||
}
|
||||
|
||||
public DualCustomBucket getClientBucket(String clientId) {
|
||||
return clientBucketMap.get(clientId);
|
||||
}
|
||||
|
||||
public Map<String, DualCustomBucket> getClientBucketMap() {
|
||||
return clientBucketMap;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 버킷 팩토리
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private DualCustomBucket makeBucket(InflowTargetVO vo) {
|
||||
if (!InflowControlManager.isInflowTarget(vo)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
LocalBucket perSecondBucket = null;
|
||||
LocalBucket thresholdBucket = null;
|
||||
|
||||
if (vo.getThresholdPerSecond() > 0) {
|
||||
perSecondBucket = Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(vo.getThresholdPerSecond(), Duration.ofSeconds(1)))
|
||||
.build();
|
||||
}
|
||||
|
||||
if (vo.getThreshold() > 0 && StringUtils.isNotBlank(vo.getThresholdTimeUnit())) {
|
||||
thresholdBucket = Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(vo.getThreshold(),
|
||||
InflowControlManager.getPeriod(vo.getThresholdTimeUnit())))
|
||||
.build();
|
||||
}
|
||||
|
||||
return new DualCustomBucket(perSecondBucket, thresholdBucket, vo);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
package com.eactive.eai.inbound.processor;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
import com.eactive.eai.common.inflow.Bucket;
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
import com.eactive.eai.common.inflow.dual.DualBucket;
|
||||
import com.eactive.eai.common.inflow.dual.DualCustomBucket;
|
||||
|
||||
/**
|
||||
* 어댑터/인터페이스 유량제어 차단 시 어느 버킷(PER_SECOND/THRESHOLD)에서 차단됐는지
|
||||
* 에러 메시지에 포함하는 RequestProcessor 확장.
|
||||
*
|
||||
* <p>DualInflowControlManager와 함께 사용:
|
||||
* <pre>
|
||||
* inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager
|
||||
* </pre>
|
||||
*
|
||||
* <p>Spring Bean 등록 후 기존 RequestProcessor 대신 이 클래스를 어댑터에 설정한다.
|
||||
*/
|
||||
public class DualRequestProcessor extends RequestProcessor {
|
||||
|
||||
@Override
|
||||
protected String checkClientInflow(Bucket bucket, String clientId) {
|
||||
if (StringUtils.isBlank(clientId)) return null;
|
||||
if (bucket instanceof DualBucket) {
|
||||
return ((DualBucket) bucket).isClientPassDetail(clientId);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected InflowTargetVO resolveClientInflowThreshold(Bucket bucket, String clientId) {
|
||||
if (bucket instanceof DualBucket) {
|
||||
return ((DualBucket) bucket).getClientInflowThreshold(clientId);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String checkAdapterInflow(Bucket bucket, String adapterGroupName) {
|
||||
if (bucket instanceof DualBucket) {
|
||||
return ((DualBucket) bucket).isAdapterPassDetail(adapterGroupName);
|
||||
}
|
||||
return super.checkAdapterInflow(bucket, adapterGroupName);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String checkInterfaceInflow(Bucket bucket, String eaiSvcCd) {
|
||||
if (bucket instanceof DualBucket) {
|
||||
return ((DualBucket) bucket).isInterfacePassDetail(eaiSvcCd);
|
||||
}
|
||||
return super.checkInterfaceInflow(bucket, eaiSvcCd);
|
||||
}
|
||||
|
||||
/**
|
||||
* 차단 원인에 따라 에러 메시지에 한도 정보를 포함.
|
||||
* <ul>
|
||||
* <li>PER_SECOND: "API 호출 한도 초과 [name: Nreq/sec]"</li>
|
||||
* <li>THRESHOLD: "API 호출 한도 초과 [name: Nreq/UNIT]"</li>
|
||||
* <li>그 외: "API 호출 한도 초과 [name]" (기존 동작)</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Override
|
||||
protected String buildInflowTargetErrorMsg(InflowTargetVO inflowTargetVO, String blockedType) {
|
||||
if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(blockedType)) {
|
||||
return String.format("API 호출 한도 초과 [%s: %dreq/sec]",
|
||||
inflowTargetVO.getName(),
|
||||
inflowTargetVO.getThresholdPerSecond());
|
||||
}
|
||||
if (DualCustomBucket.RESULT_BLOCKED_THRESHOLD.equals(blockedType)) {
|
||||
return String.format("API 호출 한도 초과 [%s: %dreq/%s]",
|
||||
inflowTargetVO.getName(),
|
||||
inflowTargetVO.getThreshold(),
|
||||
inflowTargetVO.getThresholdTimeUnit());
|
||||
}
|
||||
return super.buildInflowTargetErrorMsg(inflowTargetVO, blockedType);
|
||||
}
|
||||
}
|
||||
@@ -88,8 +88,8 @@ import java.util.Map;
|
||||
|
||||
public class RequestProcessor extends RequestProcessorSupport {
|
||||
|
||||
private static String instid = null;
|
||||
private static String serverName = null;
|
||||
protected static String instid = null;
|
||||
protected static String serverName = null;
|
||||
//private static boolean isPEAIServer = true;
|
||||
|
||||
static {
|
||||
@@ -100,9 +100,9 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
instid = eaiServerManager.getGroupInstId();
|
||||
}
|
||||
|
||||
private static boolean httpHeaderLogMode = HttpAdapterExtraLogUtil.isHttpHeaderMode();
|
||||
protected static boolean httpHeaderLogMode = HttpAdapterExtraLogUtil.isHttpHeaderMode();
|
||||
|
||||
private ProcessVO setVO(Properties prop,long msgRcvTm) throws RequestProcessorException{
|
||||
protected ProcessVO setVO(Properties prop,long msgRcvTm) throws RequestProcessorException{
|
||||
String actionName = prop.getProperty(Processor.REQUEST_ACTION);
|
||||
String adapterGroupName = prop.getProperty(Processor.ADAPTER_GROUP_NAME);
|
||||
String adapterName = prop.getProperty(Processor.ADAPTER_NAME);
|
||||
@@ -1033,13 +1033,20 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
Bucket bucket = InflowControlUtil.getBucket();
|
||||
|
||||
String blockedType = "";
|
||||
// Client 유량제어 체크 — clientId가 있는 경우에만
|
||||
String clientBlockedType = checkClientInflow(bucket, clientId);
|
||||
if (clientBlockedType != null) {
|
||||
blockedType += "C";
|
||||
}
|
||||
// Adapter 유량제어 체크
|
||||
if (!bucket.isAdapterPass(vo.getAdapterGroupName())) {
|
||||
String adapterBlockedType = checkAdapterInflow(bucket, vo.getAdapterGroupName());
|
||||
if (adapterBlockedType != null) {
|
||||
blockedType += "A";
|
||||
}
|
||||
// 그룹 우선, 없으면 인터페이스 유량제어 체크
|
||||
String groupId = bucket.getGroupIdByInterface(vo.getEaiSvcCd());
|
||||
String groupBlockedType = null;
|
||||
String interfaceBlockedType = null;
|
||||
if (groupId != null) {
|
||||
// 그룹에 속한 인터페이스 → 그룹 유량제어 체크
|
||||
groupBlockedType = bucket.isGroupPass(groupId);
|
||||
@@ -1048,7 +1055,8 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
}
|
||||
} else {
|
||||
// 그룹에 속하지 않은 인터페이스 → 인터페이스 유량제어 체크
|
||||
if (!bucket.isInterfacePass(vo.getEaiSvcCd())) {
|
||||
interfaceBlockedType = checkInterfaceInflow(bucket, vo.getEaiSvcCd());
|
||||
if (interfaceBlockedType != null) {
|
||||
blockedType += "I";
|
||||
}
|
||||
}
|
||||
@@ -1057,7 +1065,9 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
if (blockedType.length() > 0) {
|
||||
InflowTargetVO inflowTargetVO = null;
|
||||
InflowGroupVO inflowGroupVO = null;
|
||||
if (blockedType.startsWith("A")) {
|
||||
if (blockedType.startsWith("C")) {
|
||||
inflowTargetVO = resolveClientInflowThreshold(bucket, clientId);
|
||||
} else if (blockedType.startsWith("A")) {
|
||||
inflowTargetVO = bucket.getAdapterInflowThreashold(vo.getAdapterGroupName());
|
||||
} else if (blockedType.contains("G")) {
|
||||
inflowGroupVO = bucket.getGroupInflowThreshold(groupId);
|
||||
@@ -1078,7 +1088,11 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
inflowGroupVO.getThresholdTimeUnit());
|
||||
}
|
||||
} else if (inflowTargetVO != null) {
|
||||
inflowErrorMsg = String.format("API 호출 한도 초과 [%s]", inflowTargetVO.getName());
|
||||
String targetBlockedType;
|
||||
if (blockedType.startsWith("C")) targetBlockedType = clientBlockedType;
|
||||
else if (blockedType.startsWith("A")) targetBlockedType = adapterBlockedType;
|
||||
else targetBlockedType = interfaceBlockedType;
|
||||
inflowErrorMsg = buildInflowTargetErrorMsg(inflowTargetVO, targetBlockedType);
|
||||
} else {
|
||||
inflowErrorMsg = "API 호출 한도 초과";
|
||||
}
|
||||
@@ -1384,7 +1398,29 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
}
|
||||
}
|
||||
|
||||
private String getInboundErrorResponse(Properties prop, StandardMessage standardMessage, InterfaceMapper mapper
|
||||
// 클라이언트 유량제어 체크 훅 — 기본 구현은 체크 없음, DJErpRequestProcessor가 오버라이드
|
||||
protected String checkClientInflow(Bucket bucket, String clientId) {
|
||||
return null;
|
||||
}
|
||||
|
||||
protected InflowTargetVO resolveClientInflowThreshold(Bucket bucket, String clientId) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// 어댑터/인터페이스 유량제어 체크 훅 — 서브클래스에서 오버라이드하여 차단 원인(PER_SECOND/THRESHOLD)을 반환 가능
|
||||
protected String checkAdapterInflow(Bucket bucket, String adapterGroupName) {
|
||||
return bucket.isAdapterPass(adapterGroupName) ? null : "BLOCKED";
|
||||
}
|
||||
|
||||
protected String checkInterfaceInflow(Bucket bucket, String eaiSvcCd) {
|
||||
return bucket.isInterfacePass(eaiSvcCd) ? null : "BLOCKED";
|
||||
}
|
||||
|
||||
protected String buildInflowTargetErrorMsg(InflowTargetVO inflowTargetVO, String blockedType) {
|
||||
return String.format("API 호출 한도 초과 [%s]", inflowTargetVO.getName());
|
||||
}
|
||||
|
||||
protected String getInboundErrorResponse(Properties prop, StandardMessage standardMessage, InterfaceMapper mapper
|
||||
, String errCode, String errMsg, String charset) {
|
||||
String errorResponse = null;
|
||||
|
||||
@@ -1413,7 +1449,7 @@ public class RequestProcessor extends RequestProcessorSupport {
|
||||
return errorResponse;
|
||||
}
|
||||
|
||||
private String getInboundErrorResponseForStandard(StandardMessage standardMessage, InterfaceMapper mapper,
|
||||
protected String getInboundErrorResponseForStandard(StandardMessage standardMessage, InterfaceMapper mapper,
|
||||
String errCode, String errMsg, String charset, String messageType) {
|
||||
mapper.nextGuidSeq(standardMessage);
|
||||
mapper.setSendRecvDivision(standardMessage, STDMessageKeys.SEND_RECV_CD_RECV);
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
package com.eactive.eai.common.inflow;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.domain.EntityScan;
|
||||
@@ -21,21 +24,37 @@ import com.eactive.eai.data.jpa.BaseRepositoryImpl;
|
||||
@EnableJpaRepositories(repositoryBaseClass = BaseRepositoryImpl.class, basePackages = {
|
||||
"com.eactive.eai.common.inflow" })
|
||||
@EntityScan({ "com.eactive.eai.data.entity.onl.inflow" })
|
||||
@Sql({ "/com/eactive/eai/common/inflow/init_tseaifr09.sql", "/com/eactive/eai/common/inflow/init_tseaifr11.sql" })
|
||||
@Sql({
|
||||
"/com/eactive/eai/common/inflow/init_tseaifr09.sql",
|
||||
"/com/eactive/eai/common/inflow/init_tseaifr11.sql",
|
||||
"/com/eactive/eai/common/inflow/init_inflow_control_group.sql",
|
||||
"/com/eactive/eai/common/inflow/init_inflow_control_group_mapping.sql"
|
||||
})
|
||||
class InflowControlManagerTest {
|
||||
|
||||
@Autowired
|
||||
InflowControlManager inflowControlManager;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() throws LifecycleException {
|
||||
if (!inflowControlManager.isStarted()) {
|
||||
inflowControlManager.start();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void testStart() throws LifecycleException {
|
||||
// given
|
||||
|
||||
// when
|
||||
inflowControlManager.start();
|
||||
|
||||
// then
|
||||
void testStart() {
|
||||
assertTrue(inflowControlManager.isStarted());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testGetGroupBucket_existingGroup_returnsNotNull() {
|
||||
assertNotNull(inflowControlManager.getGroupBucket("test-group-001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testGetGroupBucket_unknownGroup_returnsNull() {
|
||||
assertNull(inflowControlManager.getGroupBucket("non-existent-group"));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket4j;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
class DualCustomBucketTest {
|
||||
|
||||
private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
|
||||
InflowTargetVO vo = new InflowTargetVO();
|
||||
vo.setName(name);
|
||||
vo.setThresholdPerSecond(perSec);
|
||||
vo.setThreshold(threshold);
|
||||
vo.setThresholdTimeUnit(unit);
|
||||
vo.setActivate(true);
|
||||
return vo;
|
||||
}
|
||||
|
||||
private LocalBucket freshBucket(long capacity) {
|
||||
return Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(capacity, Duration.ofSeconds(1)))
|
||||
.build();
|
||||
}
|
||||
|
||||
private LocalBucket exhaustedBucket(long capacity) {
|
||||
LocalBucket b = freshBucket(capacity);
|
||||
b.tryConsume(capacity);
|
||||
return b;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// tryConsume — 통과 케이스
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void bothBucketsNull_returnsPass() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(null, null, makeVO("A", 0, 0, null));
|
||||
assertNull(bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void onlyPerSecond_withinLimit_returnsPass() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), null, makeVO("A", 5, 0, null));
|
||||
assertNull(bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void onlyThreshold_withinLimit_returnsPass() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(null, freshBucket(100), makeVO("A", 0, 100, "MIN"));
|
||||
assertNull(bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothBuckets_bothWithinLimit_returnsPass() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), freshBucket(100), makeVO("A", 10, 100, "MIN"));
|
||||
assertNull(bucket.tryConsume());
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// tryConsume — 차단 케이스
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void onlyPerSecond_exceeded_returnsBlockedPerSecond() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), null, makeVO("A", 1, 0, null));
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void onlyThreshold_exceeded_returnsBlockedThreshold() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(null, exhaustedBucket(1), makeVO("A", 0, 1, "MIN"));
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void perSecondExceeded_thresholdRemaining_returnsBlockedPerSecond() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), freshBucket(100), makeVO("A", 1, 100, "MIN"));
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void perSecondRemaining_thresholdExceeded_returnsBlockedThreshold() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), exhaustedBucket(1), makeVO("A", 10, 1, "MIN"));
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothExceeded_returnsBlockedPerSecond() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), exhaustedBucket(1), makeVO("A", 1, 1, "MIN"));
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// tryConsume — 연속 호출
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void consecutiveConsumes_blockedAfterCapacityExhausted() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(3), null, makeVO("A", 3, 0, null));
|
||||
assertNull(bucket.tryConsume());
|
||||
assertNull(bucket.tryConsume());
|
||||
assertNull(bucket.tryConsume());
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND, bucket.tryConsume());
|
||||
}
|
||||
|
||||
@Test
|
||||
void bothBuckets_thresholdExhaustedAfterMultiplePasses() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(10), freshBucket(3), makeVO("A", 10, 3, "MIN"));
|
||||
assertNull(bucket.tryConsume());
|
||||
assertNull(bucket.tryConsume());
|
||||
assertNull(bucket.tryConsume());
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD, bucket.tryConsume());
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 가용 토큰 수 조회
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void nullBuckets_availableTokensReturnMinusOne() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(null, null, makeVO("A", 0, 0, null));
|
||||
assertEquals(-1, bucket.getPerSecondAvailableTokens());
|
||||
assertEquals(-1, bucket.getThresholdAvailableTokens());
|
||||
}
|
||||
|
||||
@Test
|
||||
void availableTokensReflectInitialCapacity() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), freshBucket(100), makeVO("A", 5, 100, "MIN"));
|
||||
assertEquals(5, bucket.getPerSecondAvailableTokens());
|
||||
assertEquals(100, bucket.getThresholdAvailableTokens());
|
||||
}
|
||||
|
||||
@Test
|
||||
void availableTokensDecrementAfterEachConsume() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(freshBucket(5), freshBucket(10), makeVO("A", 5, 10, "MIN"));
|
||||
bucket.tryConsume();
|
||||
assertEquals(4, bucket.getPerSecondAvailableTokens());
|
||||
assertEquals(9, bucket.getThresholdAvailableTokens());
|
||||
bucket.tryConsume();
|
||||
assertEquals(3, bucket.getPerSecondAvailableTokens());
|
||||
assertEquals(8, bucket.getThresholdAvailableTokens());
|
||||
}
|
||||
|
||||
@Test
|
||||
void perSecondBlocked_thresholdTokenNotConsumed() {
|
||||
DualCustomBucket bucket = new DualCustomBucket(exhaustedBucket(1), freshBucket(10), makeVO("A", 1, 10, "MIN"));
|
||||
long thresholdBefore = bucket.getThresholdAvailableTokens();
|
||||
bucket.tryConsume();
|
||||
assertEquals(thresholdBefore, bucket.getThresholdAvailableTokens());
|
||||
}
|
||||
|
||||
@Test
|
||||
void resultPassConstant_isNull() {
|
||||
assertNull(DualCustomBucket.RESULT_PASS);
|
||||
}
|
||||
}
|
||||
+344
@@ -0,0 +1,344 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket4j;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
/**
|
||||
* DualCustomBucket / DualInflowControlManager 동시성 테스트.
|
||||
*
|
||||
* <p>모든 버킷을 Duration.ofHours(1)로 생성하여 테스트 실행 중 리필이 없도록 한다.
|
||||
* CountDownLatch로 모든 스레드를 동시 출발시켜 race window를 최대화한다.
|
||||
*/
|
||||
class DualInflowControlConcurrencyTest {
|
||||
|
||||
private static final int THREAD_COUNT = 20;
|
||||
private static final int TRIES_PER_THREAD = 50;
|
||||
private static final int TOTAL_TRIES = THREAD_COUNT * TRIES_PER_THREAD; // 1000
|
||||
|
||||
private DualInflowControlManager manager;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
manager = new DualInflowControlManager();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
|
||||
InflowTargetVO vo = new InflowTargetVO();
|
||||
vo.setName(name);
|
||||
vo.setThresholdPerSecond(perSec);
|
||||
vo.setThreshold(threshold);
|
||||
vo.setThresholdTimeUnit(unit);
|
||||
vo.setActivate(true);
|
||||
return vo;
|
||||
}
|
||||
|
||||
private LocalBucket stableBucket(long capacity) {
|
||||
return Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(capacity, Duration.ofHours(1)))
|
||||
.build();
|
||||
}
|
||||
|
||||
private boolean runConcurrently(int threads, int triesPerThread, Runnable work)
|
||||
throws InterruptedException {
|
||||
CountDownLatch ready = new CountDownLatch(threads);
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
CountDownLatch done = new CountDownLatch(threads);
|
||||
ExecutorService executor = Executors.newFixedThreadPool(threads);
|
||||
|
||||
for (int i = 0; i < threads; i++) {
|
||||
executor.submit(() -> {
|
||||
ready.countDown();
|
||||
try {
|
||||
start.await();
|
||||
for (int j = 0; j < triesPerThread; j++) {
|
||||
work.run();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
} finally {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
ready.await();
|
||||
start.countDown();
|
||||
boolean completed = done.await(30, TimeUnit.SECONDS);
|
||||
executor.shutdown();
|
||||
return completed;
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// DualCustomBucket 동시성 테스트
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void perSecondOnly_totalPassEqualsCapacity() throws InterruptedException {
|
||||
long capacity = 200;
|
||||
DualCustomBucket bucket = new DualCustomBucket(
|
||||
stableBucket(capacity), null, makeVO("A", capacity, 0, null));
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockedPerSecond = new AtomicInteger();
|
||||
AtomicInteger blockedThreshold = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = bucket.tryConsume();
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result)) blockedPerSecond.incrementAndGet();
|
||||
else blockedThreshold.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
|
||||
assertEquals(capacity, passCount.get());
|
||||
assertEquals(0, blockedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void thresholdOnly_totalPassEqualsCapacity() throws InterruptedException {
|
||||
long capacity = 200;
|
||||
DualCustomBucket bucket = new DualCustomBucket(
|
||||
null, stableBucket(capacity), makeVO("A", 0, capacity, "HOUR"));
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockedThreshold = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = bucket.tryConsume();
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else if (DualCustomBucket.RESULT_BLOCKED_THRESHOLD.equals(result)) blockedThreshold.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockedThreshold.get());
|
||||
assertEquals(capacity, passCount.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void dualBuckets_thresholdIsBottleneck_totalPassEqualsThreshold() throws InterruptedException {
|
||||
long perSecCapacity = 2000;
|
||||
long threshCapacity = 200;
|
||||
|
||||
DualCustomBucket bucket = new DualCustomBucket(
|
||||
stableBucket(perSecCapacity), stableBucket(threshCapacity),
|
||||
makeVO("A", perSecCapacity, threshCapacity, "HOUR"));
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockedPerSecond = new AtomicInteger();
|
||||
AtomicInteger blockedThreshold = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = bucket.tryConsume();
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result)) blockedPerSecond.incrementAndGet();
|
||||
else blockedThreshold.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
|
||||
assertEquals(threshCapacity, passCount.get());
|
||||
assertEquals(0, blockedPerSecond.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void dualBuckets_perSecondIsBottleneck_noThresholdTokenLeak() throws InterruptedException {
|
||||
long perSecCapacity = 100;
|
||||
long threshCapacity = 2000;
|
||||
|
||||
DualCustomBucket bucket = new DualCustomBucket(
|
||||
stableBucket(perSecCapacity), stableBucket(threshCapacity),
|
||||
makeVO("A", perSecCapacity, threshCapacity, "HOUR"));
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockedPerSecond = new AtomicInteger();
|
||||
AtomicInteger blockedThreshold = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = bucket.tryConsume();
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else if (DualCustomBucket.RESULT_BLOCKED_PER_SECOND.equals(result)) blockedPerSecond.incrementAndGet();
|
||||
else blockedThreshold.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockedPerSecond.get() + blockedThreshold.get());
|
||||
assertEquals(perSecCapacity, passCount.get());
|
||||
assertEquals(0, blockedThreshold.get());
|
||||
assertEquals(perSecCapacity, threshCapacity - bucket.getThresholdAvailableTokens());
|
||||
}
|
||||
|
||||
@Test
|
||||
void availableTokensConsistentAfterConcurrentConsume() throws InterruptedException {
|
||||
long capacity = 300;
|
||||
DualCustomBucket bucket = new DualCustomBucket(
|
||||
stableBucket(capacity), null, makeVO("A", capacity, 0, null));
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
if (bucket.tryConsume() == null) passCount.incrementAndGet();
|
||||
});
|
||||
|
||||
long remaining = bucket.getPerSecondAvailableTokens();
|
||||
assertEquals(capacity, passCount.get() + remaining);
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// DualInflowControlManager 동시성 테스트
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void manager_isAdapterPassDetail_concurrent_totalPassEqualsCapacity() throws InterruptedException {
|
||||
long capacity = 200;
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", new DualCustomBucket(
|
||||
stableBucket(capacity), null, makeVO("ADAPTER_A", capacity, 0, null)));
|
||||
ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockCount = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = manager.isAdapterPassDetail("ADAPTER_A");
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else blockCount.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
|
||||
assertEquals(capacity, passCount.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void manager_isInterfacePassDetail_concurrent_totalPassEqualsCapacity() throws InterruptedException {
|
||||
long capacity = 200;
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", new DualCustomBucket(
|
||||
null, stableBucket(capacity), makeVO("IF_001", 0, capacity, "HOUR")));
|
||||
ReflectionTestUtils.setField(manager, "interfaceBucketMap", map);
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockCount = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
String result = manager.isInterfacePassDetail("IF_001");
|
||||
if (result == null) passCount.incrementAndGet();
|
||||
else blockCount.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
|
||||
assertEquals(capacity, passCount.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void manager_volatileMapSwap_noConcurrentException() throws InterruptedException {
|
||||
String adapterName = "ADAPTER_A";
|
||||
int readerCount = THREAD_COUNT;
|
||||
int swapCount = 200;
|
||||
|
||||
Map<String, DualCustomBucket> initialMap = new ConcurrentHashMap<>();
|
||||
initialMap.put(adapterName, new DualCustomBucket(
|
||||
stableBucket(10_000), null, makeVO(adapterName, 10_000, 0, null)));
|
||||
ReflectionTestUtils.setField(manager, "adapterBucketMap", initialMap);
|
||||
|
||||
AtomicReference<Throwable> error = new AtomicReference<>();
|
||||
CountDownLatch ready = new CountDownLatch(readerCount + 1);
|
||||
CountDownLatch start = new CountDownLatch(1);
|
||||
CountDownLatch done = new CountDownLatch(readerCount + 1);
|
||||
ExecutorService executor = Executors.newFixedThreadPool(readerCount + 1);
|
||||
|
||||
for (int i = 0; i < readerCount; i++) {
|
||||
executor.submit(() -> {
|
||||
ready.countDown();
|
||||
try {
|
||||
start.await();
|
||||
for (int j = 0; j < TRIES_PER_THREAD; j++) {
|
||||
manager.isAdapterPassDetail(adapterName);
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
error.compareAndSet(null, t);
|
||||
} finally {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
executor.submit(() -> {
|
||||
ready.countDown();
|
||||
try {
|
||||
start.await();
|
||||
for (int i = 0; i < swapCount; i++) {
|
||||
Map<String, DualCustomBucket> newMap = new ConcurrentHashMap<>();
|
||||
newMap.put(adapterName, new DualCustomBucket(
|
||||
stableBucket(10_000), null, makeVO(adapterName, 10_000, 0, null)));
|
||||
ReflectionTestUtils.setField(manager, "adapterBucketMap", newMap);
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
error.compareAndSet(null, t);
|
||||
} finally {
|
||||
done.countDown();
|
||||
}
|
||||
});
|
||||
|
||||
ready.await();
|
||||
start.countDown();
|
||||
boolean completed = done.await(30, TimeUnit.SECONDS);
|
||||
executor.shutdown();
|
||||
|
||||
assertTrue(completed, "테스트가 타임아웃되었습니다");
|
||||
assertNull(error.get(), error.get() != null ? "예외 발생: " + error.get() : null);
|
||||
}
|
||||
|
||||
@Test
|
||||
void manager_mixedPassAndDetailCalls_totalPassConsistent() throws InterruptedException {
|
||||
long capacity = 300;
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", new DualCustomBucket(
|
||||
stableBucket(capacity), null, makeVO("ADAPTER_A", capacity, 0, null)));
|
||||
ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
|
||||
|
||||
AtomicInteger passCount = new AtomicInteger();
|
||||
AtomicInteger blockCount = new AtomicInteger();
|
||||
AtomicInteger callIndex = new AtomicInteger();
|
||||
|
||||
boolean done = runConcurrently(THREAD_COUNT, TRIES_PER_THREAD, () -> {
|
||||
boolean useDetail = (callIndex.getAndIncrement() % 2 == 0);
|
||||
boolean passed;
|
||||
if (useDetail) {
|
||||
passed = (manager.isAdapterPassDetail("ADAPTER_A") == null);
|
||||
} else {
|
||||
passed = manager.isAdapterPass("ADAPTER_A");
|
||||
}
|
||||
if (passed) passCount.incrementAndGet();
|
||||
else blockCount.incrementAndGet();
|
||||
});
|
||||
|
||||
assertTrue(done, "테스트가 타임아웃되었습니다");
|
||||
assertEquals(TOTAL_TRIES, passCount.get() + blockCount.get());
|
||||
assertEquals(capacity, passCount.get());
|
||||
}
|
||||
}
|
||||
+318
@@ -0,0 +1,318 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import com.eactive.eai.common.inflow.CustomGroupBucket;
|
||||
import com.eactive.eai.common.inflow.InflowGroupVO;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket4j;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
/**
|
||||
* DualInflowControlManager 그룹 메트릭 단위 테스트.
|
||||
*
|
||||
* <p>start() 없이 groupBucketList를 ReflectionTestUtils로 직접 주입하여
|
||||
* isGroupPass() 카운팅과 getGroupMetrics / reset 메서드를 검증한다.
|
||||
*/
|
||||
class DualInflowControlManagerMetricsTest {
|
||||
|
||||
private DualInflowControlManager manager;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
manager = new DualInflowControlManager();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private InflowGroupVO makeGroupVO(String groupId, String groupName) {
|
||||
InflowGroupVO vo = new InflowGroupVO();
|
||||
vo.setGroupId(groupId);
|
||||
vo.setGroupName(groupName);
|
||||
vo.setActivate(true);
|
||||
return vo;
|
||||
}
|
||||
|
||||
private LocalBucket stableBucket(long capacity) {
|
||||
return Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(capacity, Duration.ofHours(1)))
|
||||
.build();
|
||||
}
|
||||
|
||||
private LocalBucket exhaustedBucket(long capacity) {
|
||||
LocalBucket b = stableBucket(capacity);
|
||||
b.tryConsume(capacity);
|
||||
return b;
|
||||
}
|
||||
|
||||
private CustomGroupBucket passBucket(String groupId) {
|
||||
return new CustomGroupBucket(stableBucket(100), null, makeGroupVO(groupId, groupId + "-name"));
|
||||
}
|
||||
|
||||
private CustomGroupBucket blockedPerSecondBucket(String groupId) {
|
||||
return new CustomGroupBucket(exhaustedBucket(1), null, makeGroupVO(groupId, groupId + "-name"));
|
||||
}
|
||||
|
||||
private CustomGroupBucket blockedThresholdBucket(String groupId) {
|
||||
return new CustomGroupBucket(stableBucket(100), exhaustedBucket(1), makeGroupVO(groupId, groupId + "-name"));
|
||||
}
|
||||
|
||||
private void injectGroupBucketList(Map<String, CustomGroupBucket> map) {
|
||||
ReflectionTestUtils.setField(manager, "groupBucketList", map);
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// TargetMetrics 내부 클래스
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void targetMetrics_initialState_allCountersZero() {
|
||||
DualInflowControlManager.TargetMetrics m = new DualInflowControlManager.TargetMetrics();
|
||||
|
||||
assertEquals(0, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
assertTrue(m.lastResetTime > 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void targetMetrics_reset_countersZeroAndResetTimeUpdated() throws InterruptedException {
|
||||
DualInflowControlManager.TargetMetrics m = new DualInflowControlManager.TargetMetrics();
|
||||
m.allowed.set(50);
|
||||
m.rejectedPerSecond.set(10);
|
||||
m.rejectedThreshold.set(5);
|
||||
long before = System.currentTimeMillis();
|
||||
|
||||
Thread.sleep(1);
|
||||
m.reset();
|
||||
|
||||
assertEquals(0, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
assertTrue(m.lastResetTime >= before);
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// isGroupPass() — 카운터 증가
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void isGroupPass_groupNotInBucketList_incrementsAllowedAndReturnsNull() {
|
||||
String result = manager.isGroupPass("UNKNOWN_G");
|
||||
|
||||
assertNull(result);
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("UNKNOWN_G");
|
||||
assertNotNull(m);
|
||||
assertEquals(1, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isGroupPass_pass_incrementsAllowed() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("G1", passBucket("G1"));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
String result = manager.isGroupPass("G1");
|
||||
|
||||
assertNull(result);
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
|
||||
assertEquals(1, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isGroupPass_blockedPerSecond_incrementsRejectedPerSecond() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("G1", blockedPerSecondBucket("G1"));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
String result = manager.isGroupPass("G1");
|
||||
|
||||
assertEquals(CustomGroupBucket.RESULT_BLOCKED_PER_SECOND, result);
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
|
||||
assertEquals(0, m.allowed.get());
|
||||
assertEquals(1, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isGroupPass_blockedThreshold_incrementsRejectedThreshold() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("G1", blockedThresholdBucket("G1"));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
String result = manager.isGroupPass("G1");
|
||||
|
||||
assertEquals(CustomGroupBucket.RESULT_BLOCKED_THRESHOLD, result);
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
|
||||
assertEquals(0, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(1, m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isGroupPass_multipleCalls_countersAccumulate() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
LocalBucket b = stableBucket(5);
|
||||
map.put("G1", new CustomGroupBucket(b, null, makeGroupVO("G1", "G1-name")));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
for (int i = 0; i < 8; i++) {
|
||||
manager.isGroupPass("G1");
|
||||
}
|
||||
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
|
||||
assertEquals(5, m.allowed.get());
|
||||
assertEquals(3, m.rejectedPerSecond.get());
|
||||
assertEquals(8, m.allowed.get() + m.rejectedPerSecond.get() + m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void isGroupPass_differentGroups_independentMetrics() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("G1", passBucket("G1"));
|
||||
map.put("G2", blockedPerSecondBucket("G2"));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G2");
|
||||
|
||||
assertEquals(2, manager.getGroupMetrics("G1").allowed.get());
|
||||
assertEquals(0, manager.getGroupMetrics("G1").rejectedPerSecond.get());
|
||||
assertEquals(0, manager.getGroupMetrics("G2").allowed.get());
|
||||
assertEquals(1, manager.getGroupMetrics("G2").rejectedPerSecond.get());
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// getGroupMetrics / getAllGroupMetrics
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void getGroupMetrics_beforeAnyCall_returnsNull() {
|
||||
assertNull(manager.getGroupMetrics("G1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getGroupMetrics_afterCall_returnsNonNull() {
|
||||
manager.isGroupPass("G1");
|
||||
assertNotNull(manager.getGroupMetrics("G1"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getAllGroupMetrics_empty_returnsEmptyMap() {
|
||||
assertTrue(manager.getAllGroupMetrics().isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
void getAllGroupMetrics_afterMultipleGroups_containsAll() {
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G2");
|
||||
manager.isGroupPass("G3");
|
||||
|
||||
Map<String, DualInflowControlManager.TargetMetrics> all = manager.getAllGroupMetrics();
|
||||
assertEquals(3, all.size());
|
||||
assertTrue(all.containsKey("G1"));
|
||||
assertTrue(all.containsKey("G2"));
|
||||
assertTrue(all.containsKey("G3"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getAllGroupMetrics_returnsUnmodifiableView() {
|
||||
manager.isGroupPass("G1");
|
||||
|
||||
Map<String, DualInflowControlManager.TargetMetrics> all = manager.getAllGroupMetrics();
|
||||
assertThrows(UnsupportedOperationException.class, () -> all.put("NEW", null));
|
||||
}
|
||||
|
||||
// =========================================================================
|
||||
// resetGroupMetrics / resetAllGroupMetrics
|
||||
// =========================================================================
|
||||
|
||||
@Test
|
||||
void resetGroupMetrics_resetsTargetGroup() {
|
||||
Map<String, CustomGroupBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("G1", passBucket("G1"));
|
||||
injectGroupBucketList(map);
|
||||
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G1");
|
||||
assertEquals(2, manager.getGroupMetrics("G1").allowed.get());
|
||||
|
||||
manager.resetGroupMetrics("G1");
|
||||
|
||||
DualInflowControlManager.TargetMetrics m = manager.getGroupMetrics("G1");
|
||||
assertEquals(0, m.allowed.get());
|
||||
assertEquals(0, m.rejectedPerSecond.get());
|
||||
assertEquals(0, m.rejectedThreshold.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetGroupMetrics_updatesLastResetTime() throws InterruptedException {
|
||||
manager.isGroupPass("G1");
|
||||
long timeBefore = manager.getGroupMetrics("G1").lastResetTime;
|
||||
|
||||
Thread.sleep(1);
|
||||
manager.resetGroupMetrics("G1");
|
||||
|
||||
assertTrue(manager.getGroupMetrics("G1").lastResetTime >= timeBefore);
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetGroupMetrics_nonExistentGroup_noException() {
|
||||
assertDoesNotThrow(() -> manager.resetGroupMetrics("NON_EXISTENT"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetGroupMetrics_doesNotAffectOtherGroups() {
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G2");
|
||||
|
||||
manager.resetGroupMetrics("G1");
|
||||
|
||||
assertEquals(0, manager.getGroupMetrics("G1").allowed.get());
|
||||
assertEquals(1, manager.getGroupMetrics("G2").allowed.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetAllGroupMetrics_resetsAllGroups() {
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G2");
|
||||
|
||||
manager.resetAllGroupMetrics();
|
||||
|
||||
assertEquals(0, manager.getGroupMetrics("G1").allowed.get());
|
||||
assertEquals(0, manager.getGroupMetrics("G2").allowed.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetAllGroupMetrics_emptyMap_noException() {
|
||||
assertDoesNotThrow(() -> manager.resetAllGroupMetrics());
|
||||
}
|
||||
|
||||
@Test
|
||||
void resetAllGroupMetrics_preservesKeys() {
|
||||
manager.isGroupPass("G1");
|
||||
manager.isGroupPass("G2");
|
||||
|
||||
manager.resetAllGroupMetrics();
|
||||
|
||||
assertNotNull(manager.getGroupMetrics("G1"));
|
||||
assertNotNull(manager.getGroupMetrics("G2"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
package com.eactive.eai.common.inflow.dual;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket4j;
|
||||
import io.github.bucket4j.local.LocalBucket;
|
||||
|
||||
/**
|
||||
* DualInflowControlManager 단위 테스트.
|
||||
*
|
||||
* <p>start() 호출 없이 adapterBucketMap / interfaceBucketMap을
|
||||
* ReflectionTestUtils로 직접 주입하여 detail 메서드만 검증한다.
|
||||
*/
|
||||
class DualInflowControlManagerTest {
|
||||
|
||||
private DualInflowControlManager manager;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
manager = new DualInflowControlManager();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private InflowTargetVO makeVO(String name) {
|
||||
InflowTargetVO vo = new InflowTargetVO();
|
||||
vo.setName(name);
|
||||
vo.setThresholdPerSecond(10);
|
||||
vo.setThreshold(100);
|
||||
vo.setThresholdTimeUnit("MIN");
|
||||
vo.setActivate(true);
|
||||
return vo;
|
||||
}
|
||||
|
||||
private LocalBucket freshBucket(long capacity) {
|
||||
return Bucket4j.builder()
|
||||
.addLimit(Bandwidth.simple(capacity, Duration.ofSeconds(1)))
|
||||
.build();
|
||||
}
|
||||
|
||||
private LocalBucket exhaustedBucket(long capacity) {
|
||||
LocalBucket b = freshBucket(capacity);
|
||||
b.tryConsume(capacity);
|
||||
return b;
|
||||
}
|
||||
|
||||
private DualCustomBucket passBucket(String name) {
|
||||
return new DualCustomBucket(freshBucket(10), freshBucket(100), makeVO(name));
|
||||
}
|
||||
|
||||
private DualCustomBucket blockedPerSecondBucket(String name) {
|
||||
return new DualCustomBucket(exhaustedBucket(1), freshBucket(100), makeVO(name));
|
||||
}
|
||||
|
||||
private DualCustomBucket blockedThresholdBucket(String name) {
|
||||
return new DualCustomBucket(freshBucket(10), exhaustedBucket(1), makeVO(name));
|
||||
}
|
||||
|
||||
private void injectAdapterMap(Map<String, DualCustomBucket> map) {
|
||||
ReflectionTestUtils.setField(manager, "adapterBucketMap", map);
|
||||
}
|
||||
|
||||
private void injectInterfaceMap(Map<String, DualCustomBucket> map) {
|
||||
ReflectionTestUtils.setField(manager, "interfaceBucketMap", map);
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// isAdapterPassDetail
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void isAdapterPassDetail_adapterNotInMap_returnsNull() {
|
||||
assertNull(manager.isAdapterPassDetail("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isAdapterPassDetail_adapterInMap_pass_returnsNull() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", passBucket("ADAPTER_A"));
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertNull(manager.isAdapterPassDetail("ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isAdapterPassDetail_adapterInMap_blockedPerSecond() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", blockedPerSecondBucket("ADAPTER_A"));
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
|
||||
manager.isAdapterPassDetail("ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isAdapterPassDetail_adapterInMap_blockedThreshold() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", blockedThresholdBucket("ADAPTER_A"));
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
|
||||
manager.isAdapterPassDetail("ADAPTER_A"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// isInterfacePassDetail
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void isInterfacePassDetail_interfaceNotInMap_returnsNull() {
|
||||
assertNull(manager.isInterfacePassDetail("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInterfacePassDetail_interfaceInMap_pass_returnsNull() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", passBucket("IF_001"));
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertNull(manager.isInterfacePassDetail("IF_001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInterfacePassDetail_interfaceInMap_blockedPerSecond() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", blockedPerSecondBucket("IF_001"));
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
|
||||
manager.isInterfacePassDetail("IF_001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInterfacePassDetail_interfaceInMap_blockedThreshold() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", blockedThresholdBucket("IF_001"));
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
|
||||
manager.isInterfacePassDetail("IF_001"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// isAdapterPass — boolean 래퍼
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void isAdapterPass_notInMap_returnsTrue() {
|
||||
assertTrue(manager.isAdapterPass("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isAdapterPass_pass_returnsTrue() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", passBucket("ADAPTER_A"));
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertTrue(manager.isAdapterPass("ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isAdapterPass_blocked_returnsFalse() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("ADAPTER_A", blockedPerSecondBucket("ADAPTER_A"));
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertFalse(manager.isAdapterPass("ADAPTER_A"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// isInterfacePass — boolean 래퍼
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void isInterfacePass_notInMap_returnsTrue() {
|
||||
assertTrue(manager.isInterfacePass("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInterfacePass_pass_returnsTrue() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", passBucket("IF_001"));
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertTrue(manager.isInterfacePass("IF_001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void isInterfacePass_blocked_returnsFalse() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
map.put("IF_001", blockedThresholdBucket("IF_001"));
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertFalse(manager.isInterfacePass("IF_001"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// 모니터링 접근자
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void getAdapterBucket_returnsCorrectBucket() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
DualCustomBucket expected = passBucket("ADAPTER_A");
|
||||
map.put("ADAPTER_A", expected);
|
||||
injectAdapterMap(map);
|
||||
|
||||
assertSame(expected, manager.getAdapterBucket("ADAPTER_A"));
|
||||
assertNull(manager.getAdapterBucket("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getInterfaceBucket_returnsCorrectBucket() {
|
||||
Map<String, DualCustomBucket> map = new ConcurrentHashMap<>();
|
||||
DualCustomBucket expected = passBucket("IF_001");
|
||||
map.put("IF_001", expected);
|
||||
injectInterfaceMap(map);
|
||||
|
||||
assertSame(expected, manager.getInterfaceBucket("IF_001"));
|
||||
assertNull(manager.getInterfaceBucket("UNKNOWN"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void getAdapterBucketMap_containsInjectedEntries() {
|
||||
Map<String, DualCustomBucket> injected = new ConcurrentHashMap<>();
|
||||
injected.put("ADAPTER_A", passBucket("ADAPTER_A"));
|
||||
injectAdapterMap(injected);
|
||||
|
||||
assertTrue(manager.getAdapterBucketMap().containsKey("ADAPTER_A"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
package com.eactive.eai.inbound.processor;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import com.eactive.eai.common.inflow.Bucket;
|
||||
import com.eactive.eai.common.inflow.InflowTargetVO;
|
||||
import com.eactive.eai.common.inflow.dual.DualBucket;
|
||||
import com.eactive.eai.common.inflow.dual.DualCustomBucket;
|
||||
import com.eactive.eai.common.server.EAIServerManager;
|
||||
import com.eactive.eai.common.util.ApplicationContextProvider;
|
||||
|
||||
/**
|
||||
* DualRequestProcessor 단위 테스트.
|
||||
*
|
||||
* <p>RequestProcessor 클래스는 static 초기화 블록에서 EAIServerManager.getInstance()를 호출한다.
|
||||
* @BeforeAll에서 mock ApplicationContext를 주입한 뒤 DualRequestProcessor를 최초 로딩시킨다.
|
||||
*
|
||||
* <p>주의: DualRequestProcessor를 필드 타입으로 선언하면 테스트 클래스 로딩 시 함께 로딩되어
|
||||
* @BeforeAll 실행 전에 NPE가 발생한다. 반드시 메서드 본문 내에서만 참조해야 한다.
|
||||
*/
|
||||
class DualRequestProcessorTest {
|
||||
|
||||
@BeforeAll
|
||||
static void setupMockApplicationContext() {
|
||||
ApplicationContext mockContext = mock(ApplicationContext.class);
|
||||
EAIServerManager mockServerManager = mock(EAIServerManager.class);
|
||||
when(mockContext.getBean(EAIServerManager.class)).thenReturn(mockServerManager);
|
||||
when(mockServerManager.getLocalServerName()).thenReturn("TEST-SERVER");
|
||||
when(mockServerManager.getGroupInstId()).thenReturn("TEST-INST");
|
||||
|
||||
ReflectionTestUtils.setField(ApplicationContextProvider.class, "context", mockContext);
|
||||
}
|
||||
|
||||
private InflowTargetVO makeVO(String name, long perSec, long threshold, String unit) {
|
||||
InflowTargetVO vo = new InflowTargetVO();
|
||||
vo.setName(name);
|
||||
vo.setThresholdPerSecond(perSec);
|
||||
vo.setThreshold(threshold);
|
||||
vo.setThresholdTimeUnit(unit);
|
||||
vo.setActivate(true);
|
||||
return vo;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// checkAdapterInflow
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void checkAdapterInflow_dualBucket_pass_returnsNull() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
DualBucket mockBucket = mock(DualBucket.class);
|
||||
when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(null);
|
||||
|
||||
assertNull(processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
|
||||
verify(mockBucket).isAdapterPassDetail("ADAPTER_A");
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkAdapterInflow_dualBucket_blockedPerSecond() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
DualBucket mockBucket = mock(DualBucket.class);
|
||||
when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
|
||||
processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkAdapterInflow_dualBucket_blockedThreshold() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
DualBucket mockBucket = mock(DualBucket.class);
|
||||
when(mockBucket.isAdapterPassDetail("ADAPTER_A")).thenReturn(DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_THRESHOLD,
|
||||
processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkAdapterInflow_nonDualBucket_pass_returnsNull() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
Bucket mockBucket = mock(Bucket.class);
|
||||
when(mockBucket.isAdapterPass("ADAPTER_A")).thenReturn(true);
|
||||
|
||||
assertNull(processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkAdapterInflow_nonDualBucket_blocked_returnsBlocked() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
Bucket mockBucket = mock(Bucket.class);
|
||||
when(mockBucket.isAdapterPass("ADAPTER_A")).thenReturn(false);
|
||||
|
||||
assertEquals("BLOCKED", processor.checkAdapterInflow(mockBucket, "ADAPTER_A"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// checkInterfaceInflow
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void checkInterfaceInflow_dualBucket_pass_returnsNull() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
DualBucket mockBucket = mock(DualBucket.class);
|
||||
when(mockBucket.isInterfacePassDetail("IF_001")).thenReturn(null);
|
||||
|
||||
assertNull(processor.checkInterfaceInflow(mockBucket, "IF_001"));
|
||||
verify(mockBucket).isInterfacePassDetail("IF_001");
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkInterfaceInflow_dualBucket_blockedPerSecond() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
DualBucket mockBucket = mock(DualBucket.class);
|
||||
when(mockBucket.isInterfacePassDetail("IF_001")).thenReturn(DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
|
||||
|
||||
assertEquals(DualCustomBucket.RESULT_BLOCKED_PER_SECOND,
|
||||
processor.checkInterfaceInflow(mockBucket, "IF_001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkInterfaceInflow_nonDualBucket_pass_returnsNull() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
Bucket mockBucket = mock(Bucket.class);
|
||||
when(mockBucket.isInterfacePass("IF_001")).thenReturn(true);
|
||||
|
||||
assertNull(processor.checkInterfaceInflow(mockBucket, "IF_001"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void checkInterfaceInflow_nonDualBucket_blocked_returnsBlocked() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
Bucket mockBucket = mock(Bucket.class);
|
||||
when(mockBucket.isInterfacePass("IF_001")).thenReturn(false);
|
||||
|
||||
assertEquals("BLOCKED", processor.checkInterfaceInflow(mockBucket, "IF_001"));
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// buildInflowTargetErrorMsg
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
void buildInflowTargetErrorMsg_perSecond_includesReqPerSec() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
|
||||
|
||||
String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_PER_SECOND);
|
||||
|
||||
assertEquals("API 호출 한도 초과 [결제API: 30req/sec]", msg);
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildInflowTargetErrorMsg_threshold_includesReqPerUnit() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
|
||||
|
||||
String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
|
||||
|
||||
assertEquals("API 호출 한도 초과 [결제API: 1000req/MIN]", msg);
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildInflowTargetErrorMsg_blockedFallback_simpleFormat() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
|
||||
|
||||
String msg = processor.buildInflowTargetErrorMsg(vo, "BLOCKED");
|
||||
|
||||
assertEquals("API 호출 한도 초과 [결제API]", msg);
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildInflowTargetErrorMsg_nullBlockedType_simpleFormat() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
InflowTargetVO vo = makeVO("결제API", 30, 1000, "MIN");
|
||||
|
||||
String msg = processor.buildInflowTargetErrorMsg(vo, null);
|
||||
|
||||
assertEquals("API 호출 한도 초과 [결제API]", msg);
|
||||
}
|
||||
|
||||
@Test
|
||||
void buildInflowTargetErrorMsg_hourUnit() {
|
||||
DualRequestProcessor processor = new DualRequestProcessor();
|
||||
InflowTargetVO vo = makeVO("조회API", 10, 500, "HOUR");
|
||||
|
||||
String msg = processor.buildInflowTargetErrorMsg(vo, DualCustomBucket.RESULT_BLOCKED_THRESHOLD);
|
||||
|
||||
assertEquals("API 호출 한도 초과 [조회API: 500req/HOUR]", msg);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
INSERT INTO inflow_control_group (group_id, group_name, threshold, threshold_per_second, threshold_time_unit, use_yn) VALUES
|
||||
('test-group-001', '테스트 그룹', 100, 10, 'MIN', '1');
|
||||
@@ -0,0 +1,2 @@
|
||||
INSERT INTO inflow_control_group_mapping (interface_id, group_id, use_yn) VALUES
|
||||
('FASSFEP00000004S2', 'test-group-001', '1');
|
||||
Reference in New Issue
Block a user