/**
 * Convert a StyxService to a Guava Service.
 *
 * @param styxService
 * @return
 */
public static Service toGuavaService(StyxService styxService) {
    return new AbstractService() {
        @Override
        protected void doStart() {
            styxService.start()
                    .thenAccept(x -> notifyStarted())
                    .exceptionally(e -> {
                        notifyFailed(e);
                        return null;
                    });
        }

        @Override
        protected void doStop() {
            styxService.stop()
                    .thenAccept(x -> notifyStopped())
                    .exceptionally(e -> {
                        notifyFailed(e);
                        return null;
                    });
        }
    };
}
 

/** Verify that the name of the thread created by LeaderSelector is set correctly. */
@Test
public void testThreadName() throws Exception {
    final String expectedThreadName = "TestLeaderService";
    final SettableFuture<String> actualThreadName = SettableFuture.create();
    register(new LeaderService(_curator, PATH, "id", expectedThreadName, 1, TimeUnit.HOURS, () -> new AbstractService() {
        @Override
        protected void doStart() {
            actualThreadName.set(Thread.currentThread().getName());
            notifyStarted();
        }

        @Override
        protected void doStop() {
            notifyStopped();
        }
    })).startAsync();
    assertEquals(expectedThreadName, actualThreadName.get(1, TimeUnit.MINUTES));
}
 

private Service createServiceForPartition(final int partition) {
    return new AbstractService() {
        private ExecutorService _service;
        private CountDownLatch _latch = new CountDownLatch(1);

        @Override
        protected void doStart() {
            _service = Executors.newSingleThreadExecutor(
                    new ThreadFactoryBuilder().setNameFormat(String.format("%s-%d-%%d", _serviceName, partition)).build());
            _service.submit(() -> {
                _log.info("{}-{}: Service started", _serviceName, partition);
                _ownedPartitions.incrementAndGet();
                try {
                    while (!_service.isShutdown()) {
                        try {
                            _latch.await(5, TimeUnit.SECONDS);
                        } catch (InterruptedException e) {
                            if (!_service.isShutdown()) {
                                _log.info("{}-{}: Service thread interrupted prior to shutdown", _serviceName, partition);
                            }
                        }
                    }
                } finally {
                    _log.info("{}-{}: Service terminating", _serviceName, partition);
                    _ownedPartitions.decrementAndGet();
                }
            });
            notifyStarted();
        }

        @Override
        protected void doStop() {
            _latch.countDown();
            _service.shutdown();
            notifyStopped();
        }
    };
}