@Override
	public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
		String localNamespace = MetadataContext.LOCAL_NAMESPACE;
		String localService = MetadataContext.LOCAL_SERVICE;

		Set<Argument> arguments = rateLimitRuleArgumentResolver.getArguments(exchange, localNamespace, localService);
		long waitMs = -1;
		try {
			String path = exchange.getRequest().getURI().getPath();
			QuotaResponse quotaResponse = QuotaCheckUtils.getQuota(
					limitAPI, localNamespace, localService, 1, arguments, path);

			if (quotaResponse.getCode() == QuotaResultCode.QuotaResultLimited) {
				ServerHttpResponse response = exchange.getResponse();
				DataBuffer dataBuffer;
				if (!Objects.isNull(polarisRateLimiterLimitedFallback)) {
					response.setRawStatusCode(polarisRateLimiterLimitedFallback.rejectHttpCode());
					response.getHeaders().setContentType(polarisRateLimiterLimitedFallback.mediaType());
					dataBuffer = response.bufferFactory().allocateBuffer()
							.write(polarisRateLimiterLimitedFallback.rejectTips()
									.getBytes(polarisRateLimiterLimitedFallback.charset()));
				}
				else {
					response.setRawStatusCode(polarisRateLimitProperties.getRejectHttpCode());
					response.getHeaders().setContentType(MediaType.TEXT_HTML);
					dataBuffer = response.bufferFactory().allocateBuffer()
							.write(rejectTips.getBytes(StandardCharsets.UTF_8));
				}
				response.getHeaders()
						.add(HeaderConstant.INTERNAL_CALLEE_RET_STATUS, RetStatus.RetFlowControl.getDesc());
				if (Objects.nonNull(quotaResponse.getActiveRule())) {
					try {
						String encodedActiveRuleName = URLEncoder.encode(
								quotaResponse.getActiveRuleName(), UTF_8);
						response.getHeaders().add(HeaderConstant.INTERNAL_ACTIVE_RULE_NAME, encodedActiveRuleName);
					}
					catch (UnsupportedEncodingException e) {
						LOG.error("Cannot encode {} for header internal-callee-activerule.",
								quotaResponse.getActiveRuleName(), e);
					}
				}
				return response.writeWith(Mono.just(dataBuffer));
			}
			// Unirate
			if (quotaResponse.getCode() == QuotaResultCode.QuotaResultOk && quotaResponse.getWaitMs() > 0) {
				LOG.debug("The request of [{}] will waiting for {}ms.", path, quotaResponse.getWaitMs());
				waitMs = quotaResponse.getWaitMs();
			}
		}
		catch (Throwable t) {
			// An exception occurs in the rate limiting API call,
			// which should not affect the call of the business process.
			LOG.error("fail to invoke getQuota, service is " + localService, t);
		}

		if (waitMs > 0) {
			return Mono.delay(Duration.ofMillis(waitMs)).flatMap(e -> chain.filter(exchange));
		}
		else {
			return chain.filter(exchange);
		}
	}
@Test
	public void testFilter() {
		// Create mock WebFilterChain
		WebFilterChain webFilterChain = serverWebExchange -> Mono.empty();

		// Mock request
		MockServerHttpRequest request = MockServerHttpRequest.get("http://localhost:8080/test").build();

		quotaCheckReactiveFilter.init();

		// Pass
		MetadataContext.LOCAL_SERVICE = "TestApp1";
		ServerWebExchange testApp1Exchange = MockServerWebExchange.from(request);
		quotaCheckReactiveFilter.filter(testApp1Exchange, webFilterChain);

		// Unirate waiting 1000ms
		MetadataContext.LOCAL_SERVICE = "TestApp2";
		ServerWebExchange testApp2Exchange = MockServerWebExchange.from(request);
		long startTimestamp = System.currentTimeMillis();
		CountDownLatch countDownLatch = new CountDownLatch(1);
		quotaCheckReactiveFilter.filter(testApp2Exchange, webFilterChain).subscribe(e -> {
		}, t -> {
		}, countDownLatch::countDown);
		try {
			countDownLatch.await();
		}
		catch (InterruptedException e) {
			fail("Exception encountered.", e);
		}
		assertThat(System.currentTimeMillis() - startTimestamp).isGreaterThanOrEqualTo(1000L);

		// Rate limited
		MetadataContext.LOCAL_SERVICE = "TestApp3";
		ServerWebExchange testApp3Exchange = MockServerWebExchange.from(request);
		quotaCheckReactiveFilter.filter(testApp3Exchange, webFilterChain);
		ServerHttpResponse response = testApp3Exchange.getResponse();
		assertThat(response.getRawStatusCode()).isEqualTo(419);
		assertThat(response.getStatusCode()).isEqualTo(HttpStatus.INSUFFICIENT_SPACE_ON_RESOURCE);
		assertThat(response.getHeaders()
				.get(HeaderConstant.INTERNAL_ACTIVE_RULE_NAME)).isEqualTo(Collections.singletonList("MOCK_RULE"));

		// Exception
		MetadataContext.LOCAL_SERVICE = "TestApp4";
		ServerWebExchange testApp4Exchange = MockServerWebExchange.from(request);
		quotaCheckReactiveFilter.filter(testApp4Exchange, webFilterChain);
	}