ActiveLimitFilter implements Filter, Filter.Listener { @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0); final RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()); if (!RpcStatus.beginCount(url, methodName, max)) { long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), TIMEOUT_KEY, 0); long start = System.currentTimeMillis(); long remain = timeout; synchronized (rpcStatus) { while (!RpcStatus.beginCount(url, methodName, max)) { try { rpcStatus.wait(remain); } catch (InterruptedException e) { } long elapsed = System.currentTimeMillis() - start; remain = timeout - elapsed; if (remain <= 0) { throw new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION, "Waiting concurrent invoke timeout in client-side for service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", elapsed: " + elapsed + ", timeout: " + timeout + ". concurrent invokes: " + rpcStatus.getActive() + ". max concurrent invoke limit: " + max); } } } } invocation.put(ACTIVELIMIT_FILTER_START_TIME, System.currentTimeMillis()); return invoker.invoke(invocation); }  @Override Result invoke(Invoker<?> invoker, Invocation invocation); @Override void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation); @Override void onError(Throwable t, Invoker<?> invoker, Invocation invocation);  }
@Test public void testInvokeNoActives() { URL url = URL.valueOf("test: Invoker<ActiveLimitFilterTest> invoker = new MyInvoker<ActiveLimitFilterTest>(url); Invocation invocation = new MockInvocation(); activeLimitFilter.invoke(invoker, invocation); }
@Test public void testInvokeLessActives() { URL url = URL.valueOf("test: Invoker<ActiveLimitFilterTest> invoker = new MyInvoker<ActiveLimitFilterTest>(url); Invocation invocation = new MockInvocation(); activeLimitFilter.invoke(invoker, invocation); }
@Test public void testInvokeGreaterActives() { AtomicInteger count = new AtomicInteger(0); URL url = URL.valueOf("test: final Invoker<ActiveLimitFilterTest> invoker = new BlockMyInvoker<ActiveLimitFilterTest>(url, 100); final Invocation invocation = new MockInvocation(); final CountDownLatch latch = new CountDownLatch(1); for (int i = 0; i < 100; i++) { Thread thread = new Thread(new Runnable() { public void run() { try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } for (int i = 0; i < 100; i++) { try { activeLimitFilter.invoke(invoker, invocation); } catch (RpcException expected) { count.incrementAndGet(); } } } }); thread.start(); } latch.countDown(); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } assertNotSame(0, count.intValue()); }
@Test public void testInvokeRuntimeException() { Assertions.assertThrows(RuntimeException.class, () -> { URL url = URL.valueOf("test: Invoker<ActiveLimitFilterTest> invoker = new RuntimeExceptionInvoker(url); Invocation invocation = new MockInvocation(); RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()); int beforeExceptionActiveCount = count.getActive(); activeLimitFilter.invoke(invoker, invocation); int afterExceptionActiveCount = count.getActive(); assertEquals(beforeExceptionActiveCount, afterExceptionActiveCount, "After exception active count should be same"); }); }