RoutingFilter implements ContainerRequestFilter, ContainerResponseFilter { @Override public void filter(ContainerRequestContext request) throws IOException { if (isWhiteList(request)) { request.setProperty("whiteList", true); return; } extractShardAndBucketIdFromRequest(request); int bucketId = getBucketIdFromRequest(request); String shardId = getShardIdFromRequest(request); trace("Processing request: {} of shard={}, forwarded={}", request.getUriInfo().getAbsolutePath(), shardId, request.getHeaders().containsKey(X_FORWARDED_BY) ? request.getHeaders() .get(X_FORWARDED_BY).toString() : ""); if (hasRoutingLoop(request)) { abortResponse(request, BAD_REQUEST, "Routing loop detected"); request.setProperty("timer", errorTimer.time()); return; } if (shardId == null) { abortResponse(request, BAD_REQUEST, "Cannot find shard for bucketId=" + bucketId); request.setProperty("timer", errorTimer.time()); return; } if (blockRequest(bucketId, shardId)) { extractShardAndBucketIdWithoutCache(request); shardId = getShardIdFromRequest(request); } if (!isMyShard(shardId)) { proxyRequestToLeader(request, shardId); request.setProperty("timer", forwardTimer.time()); return; } incrementBucketCounter(bucketId); request.setProperty("requestCounter", 1); Member leader = null; try { leader = waitForLeader(shardId); } catch (InterruptedException e) { abortResponse(request, Response.Status.INTERNAL_SERVER_ERROR, "Request interrupted"); request.setProperty("timer", errorTimer.time()); return; } if (leader == null) { abortResponse(request, SERVICE_UNAVAILABLE, "No leader is available"); request.setProperty("timer", errorTimer.time()); return; } if (leader.isLocal()) { trace("Processing this request"); request.setProperty("timer", processTimer.time()); return; } proxyRequestToLeader(request, shardId); request.setProperty("timer", forwardTimer.time()); } private  RoutingFilter();   RoutingFilter(Gondola gondola, RoutingHelper routingHelper, ProxyClientProvider proxyClientProvider,
                  Map<String, RoutingService> services, ChangeLogProcessor changeLogProcessor); RoutingService getService(ContainerRequestContext request); void extractShardAndBucketIdFromRequest(ContainerRequestContext request); @Override void filter(ContainerRequestContext request); @Override void filter(ContainerRequestContext requestContext, ContainerResponseContext responseContext); static void configCheck(Config config); void blockRequestOnBuckets(Range<Integer> splitRange); void unblockRequestOnBuckets(Range<Integer> splitRange); void registerShutdownFunction(Runnable runnable); void start(); String getSiteId(); static List<RoutingFilter> getInstance(); Gondola getGondola(); ChangeLogProcessor getChangeLogProcessor(); BucketManager getBucketManager(); Map<String, List<String>> getRoutingTable(); LockManager getLockManager(); Map<Integer, AtomicInteger> getBucketRequestCounters(); static final int POLLING_TIMES; static final int RETRY_TIMES; static final int WAIT_MS; static final String X_GONDOLA_LEADER_ADDRESS; static final String X_FORWARDED_BY; static final String X_GONDOLA_ERROR; static final String X_GONDOLA_BUCKET_ID; static final String X_GONDOLA_SHARD_ID; }
@Test public void testRoutingLoop() throws Exception { when(request.getHeaderString(any())).thenReturn("foo;" + MY_APP_URI + ";bar"); router.filter(request); ArgumentCaptor<Response> response = ArgumentCaptor.forClass(Response.class); verify(request, times(1)).abortWith(response.capture()); assertEquals(response.getValue().getStatus(), Response.Status.BAD_REQUEST.getStatusCode()); }