@Test public void testRespectsTimeout() throws Exception { final long timeoutMs = 1000; OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketAlwaysClosed, nodeRunning, userExecutor); Stopwatch stopwatch = Stopwatch.createUnstarted(); stopwatch.start(); try { finder.findOpenSocketOnNode(node, 22, timeoutMs, MILLISECONDS); fail(); } catch (NoSuchElementException success) { } long timetaken = stopwatch.elapsed(MILLISECONDS); assertTrue(timetaken >= timeoutMs - EARLY_GRACE && timetaken <= timeoutMs + SLOW_GRACE, "timetaken=" + timetaken); }
@Override public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits) { ImmutableSet<HostAndPort> sockets = checkNodeHasIps(node, allowedInterfaces).transform(new Function<String, HostAndPort>() { @Override public HostAndPort apply(String from) { return HostAndPort.fromParts(from, port); } }).toSet(); long period = timeUnits.convert(1, TimeUnit.SECONDS); AtomicReference<HostAndPort> result = newReference(); Predicate<Iterable<HostAndPort>> findOrBreak = or(updateRefOnSocketOpen(result), throwISEIfNoLongerRunning(node)); logger.debug(">> blocking on sockets %s for %d %s", sockets, timeout, timeUnits); boolean passed = retryPredicate(findOrBreak, timeout, period, timeUnits).apply(sockets); if (passed) { logger.debug("<< socket %s opened", result); assert result.get() != null; return result.get(); } else { logger.warn("<< sockets %s didn't open after %d %s", sockets, timeout, timeUnits); throw new NoSuchElementException(format("could not connect to any ip address port %d on node %s", port, node)); } }
ConcurrentOpenSocketFinder implements OpenSocketFinder { @Override public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits) { ImmutableSet<HostAndPort> sockets = checkNodeHasIps(node, allowedInterfaces).transform(new Function<String, HostAndPort>() { @Override public HostAndPort apply(String from) { return HostAndPort.fromParts(from, port); } }).toSet(); long period = timeUnits.convert(1, TimeUnit.SECONDS); AtomicReference<HostAndPort> result = newReference(); Predicate<Iterable<HostAndPort>> findOrBreak = or(updateRefOnSocketOpen(result), throwISEIfNoLongerRunning(node)); logger.debug(">> blocking on sockets %s for %d %s", sockets, timeout, timeUnits); boolean passed = retryPredicate(findOrBreak, timeout, period, timeUnits).apply(sockets); if (passed) { logger.debug("<< socket %s opened", result); assert result.get() != null; return result.get(); } else { logger.warn("<< sockets %s didn't open after %d %s", sockets, timeout, timeUnits); throw new NoSuchElementException(format("could not connect to any ip address port %d on node %s", port, node)); } } }
ConcurrentOpenSocketFinder implements OpenSocketFinder { @Override public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits) { ImmutableSet<HostAndPort> sockets = checkNodeHasIps(node, allowedInterfaces).transform(new Function<String, HostAndPort>() { @Override public HostAndPort apply(String from) { return HostAndPort.fromParts(from, port); } }).toSet(); long period = timeUnits.convert(1, TimeUnit.SECONDS); AtomicReference<HostAndPort> result = newReference(); Predicate<Iterable<HostAndPort>> findOrBreak = or(updateRefOnSocketOpen(result), throwISEIfNoLongerRunning(node)); logger.debug(">> blocking on sockets %s for %d %s", sockets, timeout, timeUnits); boolean passed = retryPredicate(findOrBreak, timeout, period, timeUnits).apply(sockets); if (passed) { logger.debug("<< socket %s opened", result); assert result.get() != null; return result.get(); } else { logger.warn("<< sockets %s didn't open after %d %s", sockets, timeout, timeUnits); throw new NoSuchElementException(format("could not connect to any ip address port %d on node %s", port, node)); } } @Inject @VisibleForTesting  ConcurrentOpenSocketFinder(SocketOpen socketTester,
         @Named(TIMEOUT_NODE_RUNNING) Predicate<AtomicReference<NodeMetadata>> nodeRunning,
         @Named(PROPERTY_USER_THREADS) ListeningExecutorService userExecutor); }
ConcurrentOpenSocketFinder implements OpenSocketFinder { @Override public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits) { ImmutableSet<HostAndPort> sockets = checkNodeHasIps(node, allowedInterfaces).transform(new Function<String, HostAndPort>() { @Override public HostAndPort apply(String from) { return HostAndPort.fromParts(from, port); } }).toSet(); long period = timeUnits.convert(1, TimeUnit.SECONDS); AtomicReference<HostAndPort> result = newReference(); Predicate<Iterable<HostAndPort>> findOrBreak = or(updateRefOnSocketOpen(result), throwISEIfNoLongerRunning(node)); logger.debug(">> blocking on sockets %s for %d %s", sockets, timeout, timeUnits); boolean passed = retryPredicate(findOrBreak, timeout, period, timeUnits).apply(sockets); if (passed) { logger.debug("<< socket %s opened", result); assert result.get() != null; return result.get(); } else { logger.warn("<< sockets %s didn't open after %d %s", sockets, timeout, timeUnits); throw new NoSuchElementException(format("could not connect to any ip address port %d on node %s", port, node)); } } @Inject @VisibleForTesting  ConcurrentOpenSocketFinder(SocketOpen socketTester,
         @Named(TIMEOUT_NODE_RUNNING) Predicate<AtomicReference<NodeMetadata>> nodeRunning,
         @Named(PROPERTY_USER_THREADS) ListeningExecutorService userExecutor); @Override HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits); }
ConcurrentOpenSocketFinder implements OpenSocketFinder { @Override public HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits) { ImmutableSet<HostAndPort> sockets = checkNodeHasIps(node, allowedInterfaces).transform(new Function<String, HostAndPort>() { @Override public HostAndPort apply(String from) { return HostAndPort.fromParts(from, port); } }).toSet(); long period = timeUnits.convert(1, TimeUnit.SECONDS); AtomicReference<HostAndPort> result = newReference(); Predicate<Iterable<HostAndPort>> findOrBreak = or(updateRefOnSocketOpen(result), throwISEIfNoLongerRunning(node)); logger.debug(">> blocking on sockets %s for %d %s", sockets, timeout, timeUnits); boolean passed = retryPredicate(findOrBreak, timeout, period, timeUnits).apply(sockets); if (passed) { logger.debug("<< socket %s opened", result); assert result.get() != null; return result.get(); } else { logger.warn("<< sockets %s didn't open after %d %s", sockets, timeout, timeUnits); throw new NoSuchElementException(format("could not connect to any ip address port %d on node %s", port, node)); } } @Inject @VisibleForTesting  ConcurrentOpenSocketFinder(SocketOpen socketTester,
         @Named(TIMEOUT_NODE_RUNNING) Predicate<AtomicReference<NodeMetadata>> nodeRunning,
         @Named(PROPERTY_USER_THREADS) ListeningExecutorService userExecutor); @Override HostAndPort findOpenSocketOnNode(NodeMetadata node, final int port, long timeout, TimeUnit timeUnits);  }