From a4c9a2f2bd2b5245b62b49bcb156e9500ea1d941 Mon Sep 17 00:00:00 2001 From: curry772 Date: Mon, 27 Apr 2026 09:40:27 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20Dual=20=EC=9D=B4=EC=A4=91=20=EB=B2=84?= =?UTF-8?q?=ED=82=B7=20=EC=9C=A0=EB=9F=89=EC=A0=9C=EC=96=B4=20=ED=8C=A8?= =?UTF-8?q?=ED=82=A4=EC=A7=80=20=EC=B6=94=EA=B0=80=20(DJErp=E2=86=92Dual?= =?UTF-8?q?=20=EB=A6=AC=EB=84=A4=EC=9E=84=C2=B7=EC=9D=B4=EA=B4=80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - InflowType 열거형 추가 (ADAPTER/INTERFACE/CLIENT 타입 코드 캡슐화) - InflowControlDAO: InflowType 기반 범용 메서드 추가, 기존 메서드 위임 - RequestProcessor: private→protected 전환, 클라이언트/어댑터/인터페이스 유량제어 훅 메서드 5개 추가 (checkClientInflow 등) - dual 패키지 신규 추가 · DualCustomBucket: perSecond/threshold 이중 버킷, 차단 원인 구분 · DualBucket: isAdapterPassDetail 등 차단 원인 반환 인터페이스 · DualInflowControlManager: 그룹 메트릭 카운팅(TargetMetrics), 리로드 연동 · DualRequestProcessor: 클라이언트 유량제어 + 상세 오류 메시지 훅 구현 - 단위 테스트 5종 추가 (DualCustomBucket, Manager, Metrics, RequestProcessor, Concurrency) Co-Authored-By: Claude Sonnet 4.6 --- .../eai/common/inflow/InflowControlDAO.java | 24 +- .../eactive/eai/common/inflow/InflowType.java | 23 ++ .../eai/common/inflow/dual/DualBucket.java | 37 ++ .../common/inflow/dual/DualCustomBucket.java | 58 +++ .../inflow/dual/DualInflowControlManager.java | 313 ++++++++++++++++ .../inflow/dual/DualRequestProcessor.java | 80 ++++ .../inbound/processor/RequestProcessor.java | 56 ++- .../inflow/dual/DualCustomBucketTest.java | 164 +++++++++ .../DualInflowControlConcurrencyTest.java | 344 ++++++++++++++++++ .../DualInflowControlManagerMetricsTest.java | 318 ++++++++++++++++ .../dual/DualInflowControlManagerTest.java | 244 +++++++++++++ .../inflow/dual/DualRequestProcessorTest.java | 195 ++++++++++ 12 files changed, 1842 insertions(+), 14 deletions(-) create mode 100644 src/main/java/com/eactive/eai/common/inflow/InflowType.java create mode 100644 src/main/java/com/eactive/eai/common/inflow/dual/DualBucket.java create mode 100644 src/main/java/com/eactive/eai/common/inflow/dual/DualCustomBucket.java create mode 100644 src/main/java/com/eactive/eai/common/inflow/dual/DualInflowControlManager.java create mode 100644 src/main/java/com/eactive/eai/common/inflow/dual/DualRequestProcessor.java create mode 100644 src/test/java/com/eactive/eai/common/inflow/dual/DualCustomBucketTest.java create mode 100644 src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlConcurrencyTest.java create mode 100644 src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerMetricsTest.java create mode 100644 src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerTest.java create mode 100644 src/test/java/com/eactive/eai/common/inflow/dual/DualRequestProcessorTest.java diff --git a/src/main/java/com/eactive/eai/common/inflow/InflowControlDAO.java b/src/main/java/com/eactive/eai/common/inflow/InflowControlDAO.java index 9cbd12e..f8f8838 100644 --- a/src/main/java/com/eactive/eai/common/inflow/InflowControlDAO.java +++ b/src/main/java/com/eactive/eai/common/inflow/InflowControlDAO.java @@ -41,20 +41,36 @@ public class InflowControlDAO extends BaseDAO { @Autowired private InflowControlGroupMappingLoader inflowControlGroupMappingLoader; + // ------------------------------------------------------------------------- + // Enum 기반 범용 메서드 — type 코드를 직접 노출하지 않음 + // ------------------------------------------------------------------------- + + public List getInflowList(InflowType type) throws DAOException { + return getInflowList(type.code()); + } + + public Map getInflowMap(InflowType type, String name) throws DAOException { + return getInflowMap(type.code(), name); + } + + // ------------------------------------------------------------------------- + // 기존 명명 메서드 — enum 기반 메서드로 위임 + // ------------------------------------------------------------------------- + public List getInflowAdapterList() throws DAOException { - return getInflowList("01"); + return getInflowList(InflowType.ADAPTER); } public List getInflowInterfaceList() throws DAOException { - return getInflowList("02"); + return getInflowList(InflowType.INTERFACE); } public Map getInflowTargetByAdater(String name) throws DAOException { - return getInflowMap("01", name); + return getInflowMap(InflowType.ADAPTER, name); } public Map getInflowTargetByInterface(String name) throws DAOException { - return getInflowMap("02", name); + return getInflowMap(InflowType.INTERFACE, name); } private List getInflowList(String type) throws DAOException { diff --git a/src/main/java/com/eactive/eai/common/inflow/InflowType.java b/src/main/java/com/eactive/eai/common/inflow/InflowType.java new file mode 100644 index 0000000..69e944f --- /dev/null +++ b/src/main/java/com/eactive/eai/common/inflow/InflowType.java @@ -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; + } +} diff --git a/src/main/java/com/eactive/eai/common/inflow/dual/DualBucket.java b/src/main/java/com/eactive/eai/common/inflow/dual/DualBucket.java new file mode 100644 index 0000000..7f7a47c --- /dev/null +++ b/src/main/java/com/eactive/eai/common/inflow/dual/DualBucket.java @@ -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); +} diff --git a/src/main/java/com/eactive/eai/common/inflow/dual/DualCustomBucket.java b/src/main/java/com/eactive/eai/common/inflow/dual/DualCustomBucket.java new file mode 100644 index 0000000..c16a685 --- /dev/null +++ b/src/main/java/com/eactive/eai/common/inflow/dual/DualCustomBucket.java @@ -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; } +} diff --git a/src/main/java/com/eactive/eai/common/inflow/dual/DualInflowControlManager.java b/src/main/java/com/eactive/eai/common/inflow/dual/DualInflowControlManager.java new file mode 100644 index 0000000..eac2c0c --- /dev/null +++ b/src/main/java/com/eactive/eai/common/inflow/dual/DualInflowControlManager.java @@ -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)로 관리하는 유량제어 관리자. + * + *

기존 InflowControlManager는 어댑터/인터페이스에 단일 LocalBucket(복수 Bandwidth)을 사용하여 + * 차단 시 어느 한도(초당/기간)를 초과했는지 알 수 없었다. + * 이 클래스는 그룹 버킷(CustomGroupBucket)과 동일하게 두 버킷을 분리하고 + * {@link DualBucket#isAdapterPassDetail}/{@link DualBucket#isInterfacePassDetail}로 + * 차단 원인을 반환한다. 그룹 버킷 통과 여부는 {@link TargetMetrics}로 카운팅된다. + * + *

적용 방법: INFLOW.properties에 아래 한 줄 추가. + * 

inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager
+ */ +@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 adapterBucketMap = new ConcurrentHashMap<>(); + private volatile Map interfaceBucketMap = new ConcurrentHashMap<>(); + private volatile Map clientBucketMap = new ConcurrentHashMap<>(); + + private final ConcurrentHashMap 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 loadBucketMap(InflowType type) throws Exception { + Map 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 targetMap, InflowType type, String name) throws Exception { + Map 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 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 getAdapterBucketMap() { + return adapterBucketMap; + } + + public Map getInterfaceBucketMap() { + return interfaceBucketMap; + } + + public DualCustomBucket getClientBucket(String clientId) { + return clientBucketMap.get(clientId); + } + + public Map 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); + } +} diff --git a/src/main/java/com/eactive/eai/common/inflow/dual/DualRequestProcessor.java b/src/main/java/com/eactive/eai/common/inflow/dual/DualRequestProcessor.java new file mode 100644 index 0000000..add4e9e --- /dev/null +++ b/src/main/java/com/eactive/eai/common/inflow/dual/DualRequestProcessor.java @@ -0,0 +1,80 @@ +package com.eactive.eai.common.inflow.dual; + +import org.apache.commons.lang3.StringUtils; +import org.springframework.stereotype.Component; + +import com.eactive.eai.common.inflow.Bucket; +import com.eactive.eai.common.inflow.InflowTargetVO; +import com.eactive.eai.inbound.processor.RequestProcessor; + +/** + * 어댑터/인터페이스 유량제어 차단 시 어느 버킷(PER_SECOND/THRESHOLD)에서 차단됐는지 + * 에러 메시지에 포함하는 RequestProcessor 확장. + * + *

DualInflowControlManager와 함께 사용: + * 

+ * inflow.control.bucket.className=com.eactive.eai.common.inflow.dual.DualInflowControlManager
+ *
+ * + *

Spring Bean 등록 후 기존 RequestProcessor 대신 이 클래스를 어댑터에 설정한다. + */ +@Component("dualRequestProcessor") +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); + } + + /** + * 차단 원인에 따라 에러 메시지에 한도 정보를 포함. + *

    + *
  • PER_SECOND: "API 호출 한도 초과 [name: Nreq/sec]"
    • + *
  • THRESHOLD: "API 호출 한도 초과 [name: Nreq/UNIT]"
    • + *
  • 그 외: "API 호출 한도 초과 [name]" (기존 동작)
    • + *
+ */ + @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); + } +} diff --git a/src/main/java/com/eactive/eai/inbound/processor/RequestProcessor.java b/src/main/java/com/eactive/eai/inbound/processor/RequestProcessor.java index 225ea4f..64b358d 100644 --- a/src/main/java/com/eactive/eai/inbound/processor/RequestProcessor.java +++ b/src/main/java/com/eactive/eai/inbound/processor/RequestProcessor.java @@ -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); diff --git a/src/test/java/com/eactive/eai/common/inflow/dual/DualCustomBucketTest.java b/src/test/java/com/eactive/eai/common/inflow/dual/DualCustomBucketTest.java new file mode 100644 index 0000000..187ca71 --- /dev/null +++ b/src/test/java/com/eactive/eai/common/inflow/dual/DualCustomBucketTest.java @@ -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); + } +} diff --git a/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlConcurrencyTest.java b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlConcurrencyTest.java new file mode 100644 index 0000000..56654df --- /dev/null +++ b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlConcurrencyTest.java @@ -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 동시성 테스트. + * + *

모든 버킷을 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 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 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 initialMap = new ConcurrentHashMap<>(); + initialMap.put(adapterName, new DualCustomBucket( + stableBucket(10_000), null, makeVO(adapterName, 10_000, 0, null))); + ReflectionTestUtils.setField(manager, "adapterBucketMap", initialMap); + + AtomicReference 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 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 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()); + } +} diff --git a/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerMetricsTest.java b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerMetricsTest.java new file mode 100644 index 0000000..6e59247 --- /dev/null +++ b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerMetricsTest.java @@ -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 그룹 메트릭 단위 테스트. + * + *

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 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 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 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 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 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 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 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 all = manager.getAllGroupMetrics(); + assertThrows(UnsupportedOperationException.class, () -> all.put("NEW", null)); + } + + // ========================================================================= + // resetGroupMetrics / resetAllGroupMetrics + // ========================================================================= + + @Test + void resetGroupMetrics_resetsTargetGroup() { + Map 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")); + } +} diff --git a/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerTest.java b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerTest.java new file mode 100644 index 0000000..79cafd8 --- /dev/null +++ b/src/test/java/com/eactive/eai/common/inflow/dual/DualInflowControlManagerTest.java @@ -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 단위 테스트. + * + *

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 map) { + ReflectionTestUtils.setField(manager, "adapterBucketMap", map); + } + + private void injectInterfaceMap(Map map) { + ReflectionTestUtils.setField(manager, "interfaceBucketMap", map); + } + + // ------------------------------------------------------------------------- + // isAdapterPassDetail + // ------------------------------------------------------------------------- + + @Test + void isAdapterPassDetail_adapterNotInMap_returnsNull() { + assertNull(manager.isAdapterPassDetail("UNKNOWN")); + } + + @Test + void isAdapterPassDetail_adapterInMap_pass_returnsNull() { + Map map = new ConcurrentHashMap<>(); + map.put("ADAPTER_A", passBucket("ADAPTER_A")); + injectAdapterMap(map); + + assertNull(manager.isAdapterPassDetail("ADAPTER_A")); + } + + @Test + void isAdapterPassDetail_adapterInMap_blockedPerSecond() { + Map 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 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 map = new ConcurrentHashMap<>(); + map.put("IF_001", passBucket("IF_001")); + injectInterfaceMap(map); + + assertNull(manager.isInterfacePassDetail("IF_001")); + } + + @Test + void isInterfacePassDetail_interfaceInMap_blockedPerSecond() { + Map 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 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 map = new ConcurrentHashMap<>(); + map.put("ADAPTER_A", passBucket("ADAPTER_A")); + injectAdapterMap(map); + + assertTrue(manager.isAdapterPass("ADAPTER_A")); + } + + @Test + void isAdapterPass_blocked_returnsFalse() { + Map 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 map = new ConcurrentHashMap<>(); + map.put("IF_001", passBucket("IF_001")); + injectInterfaceMap(map); + + assertTrue(manager.isInterfacePass("IF_001")); + } + + @Test + void isInterfacePass_blocked_returnsFalse() { + Map map = new ConcurrentHashMap<>(); + map.put("IF_001", blockedThresholdBucket("IF_001")); + injectInterfaceMap(map); + + assertFalse(manager.isInterfacePass("IF_001")); + } + + // ------------------------------------------------------------------------- + // 모니터링 접근자 + // ------------------------------------------------------------------------- + + @Test + void getAdapterBucket_returnsCorrectBucket() { + Map 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 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 injected = new ConcurrentHashMap<>(); + injected.put("ADAPTER_A", passBucket("ADAPTER_A")); + injectAdapterMap(injected); + + assertTrue(manager.getAdapterBucketMap().containsKey("ADAPTER_A")); + } +} diff --git a/src/test/java/com/eactive/eai/common/inflow/dual/DualRequestProcessorTest.java b/src/test/java/com/eactive/eai/common/inflow/dual/DualRequestProcessorTest.java new file mode 100644 index 0000000..7227ffb --- /dev/null +++ b/src/test/java/com/eactive/eai/common/inflow/dual/DualRequestProcessorTest.java @@ -0,0 +1,195 @@ +package com.eactive.eai.common.inflow.dual; + +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.server.EAIServerManager; +import com.eactive.eai.common.util.ApplicationContextProvider; + +/** + * DualRequestProcessor 단위 테스트. + * + *

RequestProcessor 클래스는 static 초기화 블록에서 EAIServerManager.getInstance()를 호출한다. + * @BeforeAll에서 mock ApplicationContext를 주입한 뒤 DualRequestProcessor를 최초 로딩시킨다. + * + *

주의: 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); + } +}