FeedClientImpl implements FeedClient { public static boolean waitForOperations(Instant lastResultReceived, long sleepTimeMs, long closeTimeoutMs) { if (lastResultReceived.plusMillis(closeTimeoutMs).isBefore(Instant.now())) { return false; } try { Thread.sleep(sleepTimeMs); } catch (InterruptedException e) { return false; } return true; } FeedClientImpl(SessionParams sessionParams,
                          ResultCallback resultCallback,
                          ScheduledThreadPoolExecutor timeoutExecutor,
                          Clock clock); @Override void stream(String documentId, String operationId, CharSequence documentData, Object context); @Override void close(); @Override String getStatsAsJson(); static boolean waitForOperations(Instant lastResultReceived, long sleepTimeMs, long closeTimeoutMs);  }
@Test public void testCloseWaitTimeOldTimestamp() { assertThat(FeedClientImpl.waitForOperations(Instant.now().minusSeconds(1000), sleepValueMillis, 10), is(false)); }
@Test public void testCloseWaitTimeOutInFutureStillOperations() { assertThat(FeedClientImpl.waitForOperations(Instant.now(), sleepValueMillis, 2000), is(true)); }