@Test public void testAbortsWhenNodeNotRunning() throws Exception { OpenSocketFinder finder = new ConcurrentOpenSocketFinder(socketAlwaysClosed, nodeNotRunning, userExecutor) { @Override protected <T> Predicate<T> retryPredicate(final Predicate<T> findOrBreak, long timeout, long period, TimeUnit timeUnits) { return new Predicate<T>() { @Override public boolean apply(T input) { try { findOrBreak.apply(input); fail("should have thrown IllegalStateException"); } catch (IllegalStateException e) { } return false; } }; } }; try { finder.findOpenSocketOnNode(node, 22, 2000, MILLISECONDS); fail(); } catch (NoSuchElementException e) { } }
@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);  }