ExecuteLimitFilter implements Filter, Filter.Listener { @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); int max = url.getMethodParameter(methodName, EXECUTES_KEY, 0); if (!RpcStatus.beginCount(url, methodName, max)) { throw new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION, "Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited."); } invocation.put(EXECUTE_LIMIT_FILTER_START_TIME, System.currentTimeMillis()); try { return invoker.invoke(invocation); } catch (Throwable t) { if (t instanceof RuntimeException) { throw (RuntimeException) t; } else { throw new RpcException("unexpected exception when ExecuteLimitFilter", t); } } }  @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 testNoExecuteLimitInvoke() throws Exception { Invoker invoker = Mockito.mock(Invoker.class); when(invoker.invoke(any(Invocation.class))).thenReturn(new AppResponse("result")); when(invoker.getUrl()).thenReturn(URL.valueOf("test: Invocation invocation = Mockito.mock(Invocation.class); when(invocation.getMethodName()).thenReturn("testNoExecuteLimitInvoke"); Result result = executeLimitFilter.invoke(invoker, invocation); Assertions.assertEquals("result", result.getValue()); }
@Test public void testExecuteLimitInvoke() throws Exception { Invoker invoker = Mockito.mock(Invoker.class); when(invoker.invoke(any(Invocation.class))).thenReturn(new AppResponse("result")); when(invoker.getUrl()).thenReturn(URL.valueOf("test: Invocation invocation = Mockito.mock(Invocation.class); when(invocation.getMethodName()).thenReturn("testExecuteLimitInvoke"); Result result = executeLimitFilter.invoke(invoker, invocation); Assertions.assertEquals("result", result.getValue()); }
@Test public void testMoreThanExecuteLimitInvoke() throws Exception { int maxExecute = 10; int totalExecute = 20; final AtomicInteger failed = new AtomicInteger(0); final Invocation invocation = Mockito.mock(Invocation.class); when(invocation.getMethodName()).thenReturn("testMoreThanExecuteLimitInvoke"); URL url = URL.valueOf("test: final Invoker<ExecuteLimitFilter> invoker = new BlockMyInvoker<ExecuteLimitFilter>(url, 1000); final CountDownLatch latch = new CountDownLatch(1); for (int i = 0; i < totalExecute; i++) { Thread thread = new Thread(new Runnable() { public void run() { try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } try { executeLimitFilter.invoke(invoker, invocation); } catch (RpcException expected) { failed.incrementAndGet(); } } }); thread.start(); } latch.countDown(); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Assertions.assertEquals(totalExecute - maxExecute, failed.get()); }