PrimaryHandler implements TcpHandler { @Override public Object processRequest(Object request) throws IOException { long giveup = 0; while (giveup == 0 || millisecondProvider.millisecondTime() < giveup) { TcpHandler handler; if (request instanceof PeerLocatorRequest) { handler = handlerMapping.get(PeerLocatorRequest.class); } else { handler = handlerMapping.get(request.getClass()); } if (handler != null) { return handler.processRequest(request); } if (fallbackHandler != null) { return fallbackHandler.processRequest(request); } if (giveup == 0) { int locatorWaitTime = this.locatorWaitTime; if (locatorWaitTime <= 0) { locatorWaitTime = 30; } giveup = millisecondProvider.millisecondTime() + locatorWaitTime * 1000L; } try { sleeper.sleep(1000); } catch (InterruptedException ignored) { return null; } } logger.info( "Received a location request of class {} but the handler for this is either not enabled or is not ready to process requests", request.getClass().getSimpleName()); return null; } PrimaryHandler(TcpHandler fallbackHandler, int locatorWaitTime,
      MillisecondProvider millisecondProvider,
      Sleeper sleeper); @Override synchronized void init(TcpServer tcpServer); @Override Object processRequest(Object request); @Override void shutDown(); synchronized void addHandler(Class<?> clazz, TcpHandler handler); @Override void endRequest(Object request, long startTime); @Override void endResponse(Object request, long startTime);  }
@Test public void processRequest() throws IOException { final AtomicInteger sleepCount = new AtomicInteger(); final TcpHandler fallbackHandler = null; final int locatorWaitTime = 5; final List<Long> clockTimes = Arrays.asList(0L, 1000L, 2000L, 3000L, 4000L, 5000L); PrimaryHandler primaryHandler = new PrimaryHandler(fallbackHandler, locatorWaitTime, () -> clockTimes.get(sleepCount.get()), x -> sleepCount.incrementAndGet()); final Object result = primaryHandler.processRequest(new Object()); assertThat(sleepCount.get()).isEqualTo(locatorWaitTime); assertThat(result).isNull(); }
@Test public void fallbackHandler() throws IOException { final AtomicInteger handlerInvoked = new AtomicInteger(); final TcpHandler fallbackHandler = mock(TcpHandler.class); when(fallbackHandler.processRequest(isA(Object.class))).thenAnswer(context -> { handlerInvoked.incrementAndGet(); return context.getArgument(0); }); final int locatorWaitTime = 5; PrimaryHandler primaryHandler = new PrimaryHandler(fallbackHandler, locatorWaitTime, null, null); final Object request = new Object(); final Object result = primaryHandler.processRequest(request); assertThat(result).isEqualTo(request); assertThat(handlerInvoked.get()).isEqualTo(1); }