Java 类com.google.common.util.concurrent.Service 实例源码

/**
 *
 * @param callbacks
 */
@Override
public void execute(CoinActionCallback<CurrencyCoin>... callbacks) {
    _callbacks = callbacks;
   // reinitWallet();

    final DeterministicSeed seed = createDeterministicSeed();

    _bitcoinManager.getCurrencyCoin().getWalletManager().addListener(new Service.Listener() {
        @Override
        public void terminated(Service.State from) {
            super.terminated(from);
            WalletAppKit appKit = setupWallet();

            appKit.setDownloadListener(BitcoinRecoverAction.this)
                    .setBlockingStartup(false)
                    .setUserAgent(ServiceConsts.SERVICE_APP_NAME, "0.1")
                    .restoreWalletFromSeed(seed);

            _bitcoinManager.getCurrencyCoin().setWalletManager(appKit);
            _bitcoinManager.getCurrencyCoin().getWalletManager().startAsync();
        }
    }, Executors.newSingleThreadExecutor());

    _bitcoinManager.getCurrencyCoin().getWalletManager().stopAsync();
}

public static void main(String... args) throws Exception {
    PropertyConfigurator.configure("D:/log4j.properties");

    try {
        List<Replica> members = Lists.newArrayList();
        members.add(Replica.fromString("localhost:10001"));
        members.add(Replica.fromString("localhost:10002"));
        File logDir = new File("D:/raft");
        logDir.mkdir();

        // configure the service
        RaftService raft = RaftService.newBuilder().local(Replica.fromString("localhost:10000")).members(members).logDir(logDir).timeout(300).build(new Test());

        // start this replica
        Service guavaservice = raft.startAsync();
        guavaservice.awaitRunning();

        // let's commit some things
        //            for (int i = 0; i < 10; i++) {
        //                raft.commit(new byte[] { 'O', '_', 'o' });
        //            }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public static Module module(final int configSourcePriority)
{
    return new AbstractModule()
    {
        @Override
        protected void configure()
        {
            MapBinder<Integer, DynamicConfigSource> mapBinder = MapBinder.newMapBinder(binder(), Integer.class, DynamicConfigSource.class);
            mapBinder.addBinding(configSourcePriority).to(FileDynamicConfigSource.class);
            bind(FileDynamicConfigSource.class);

            // Bind inner class as a service to ensure resource cleanup
            Multibinder.newSetBinder(binder(), Service.class).addBinding().to(FileDynamicConfigSourceService.class);
        }
    };
}

@Test
public void testCronSchedulerLifecycle() throws Exception {
  control.replay();

  Scheduler scheduler = injector.getInstance(Scheduler.class);
  assertFalse(scheduler.isStarted());

  Service cronLifecycle = boot();

  assertTrue(cronLifecycle.isRunning());
  assertTrue(scheduler.isStarted());

  cronLifecycle.stopAsync().awaitTerminated();

  assertFalse(cronLifecycle.isRunning());
  assertTrue(scheduler.isShutdown());
}

@Test
public void testJobsAreScheduled() throws Exception {
  auroraCronJob.execute(isA(JobExecutionContext.class));

  control.replay();
  final Scheduler scheduler = injector.getInstance(Scheduler.class);

  storage.write((NoResult.Quiet)
      storeProvider -> storeProvider.getCronJobStore().saveAcceptedJob(CRON_JOB));

  final CountDownLatch cronRan = new CountDownLatch(1);
  scheduler.getListenerManager().addTriggerListener(new CountDownWhenComplete(cronRan));
  Service service = boot();

  cronRan.await();

  service.stopAsync().awaitTerminated();
}

@Test
public void testBindings() throws Exception {
  Injector injector = createInjector(new AsyncModule());

  control.replay();

  Set<Service> services = injector.getInstance(
      Key.get(new TypeLiteral<Set<Service>>() { }, AppStartup.class));
  for (Service service : services) {
    service.startAsync().awaitRunning();
  }

  injector.getBindings();

  assertEquals(
      ImmutableMap.of(
          RegisterGauges.TIMEOUT_QUEUE_GAUGE, 0,
          RegisterGauges.ASYNC_TASKS_GAUGE, 0L,
          RegisterGauges.DELAY_QUEUE_GAUGE, 0),
      statsProvider.getAllValues()
  );
}

/**
 * Tests the behavior of AbstractThreadPoolService during shutdown when both a StopException and a RunException are present.
 */
@Test
public void testShutdownStopAndRunException() {
    val s = newService();

    // Stop it and verify it hasn't shut down - it should still be waiting on the runFuture.
    val stopException = new IntentionalException("stop");
    s.errorHandler(stopException);
    val runException = new IntentionalException("run");
    s.runFuture.completeExceptionally(runException);

    AssertExtensions.assertThrows(
            "Service did not fail.",
            () -> s.awaitTerminated(),
            ex -> ex instanceof IllegalStateException);
    Assert.assertEquals("Unexpected state upon failed shutdown.", Service.State.FAILED, s.state());
    Assert.assertEquals("Unexpected failure cause.", stopException, s.failureCause());
    Assert.assertEquals("Unexpected suppressed exception.", runException, s.failureCause().getSuppressed()[0]);
}

@Override
protected void startUp() throws Exception {
  Throwable failureCause = null;

  for (Service service : services) {
    try {
      service.startAndWait();
    } catch (UncheckedExecutionException e) {
      failureCause = e.getCause();
      break;
    }
  }

  if (failureCause != null) {
    // Stop all running services and then throw the failure exception
    try {
      stopAll();
    } catch (Throwable t) {
      // Ignore the stop error. Just log.
      LOG.warn("Failed when stopping all services on start failure", t);
    }

    Throwables.propagateIfPossible(failureCause, Exception.class);
    throw new RuntimeException(failureCause);
  }
}

/**
 * Tests the behavior of AbstractThreadPoolService when a normal shutdown (no errors) happens.
 */
@Test
public void testShutdownNoFailure() {
    @Cleanup
    val s = newService();

    // Stop it and verify it hasn't shut down - it should still be waiting on the runFuture.
    s.stopAsync();
    AssertExtensions.assertThrows(
            "Service stopped even though the runFuture did not complete.",
            () -> s.awaitTerminated(SHORT_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS),
            ex -> ex instanceof TimeoutException);
    Assert.assertEquals("Unexpected state while shutting down.", Service.State.STOPPING, s.state());

    // Complete the future and await normal termination.
    s.runFuture.complete(null);
    s.awaitTerminated();
    Assert.assertEquals("Unexpected state upon normal shutdown.", Service.State.TERMINATED, s.state());
}

@Override
public void running() {
  if (hasCalled(Service.State.RUNNING)) {
    return;
  }
  executor.execute(new Runnable() {
    @Override
    public void run() {
      try {
        delegate.running();
      } catch (Throwable t) {
        LOG.warn("Exception thrown from listener", t);
      }
    }
  });
}

@Override
public void stopping(final Service.State from) {
  if (hasCalled(Service.State.STOPPING)) {
    return;
  }
  executor.execute(new Runnable() {
    @Override
    public void run() {
      try {
        delegate.stopping(from);
      } catch (Throwable t) {
        LOG.warn("Exception thrown from listener", t);
      }
    }
  });
}

@Override
public void terminated(final Service.State from) {
  if (hasCalled(Service.State.TERMINATED)) {
    return;
  }
  executor.execute(new Runnable() {
    @Override
    public void run() {
      try {
        delegate.terminated(from);
      } catch (Throwable t) {
        LOG.warn("Exception thrown from listener", t);
      }
    }
  });
}

private Service createService(ZKClient zkClient, RunId runId) {
  return new AbstractTwillService(zkClient, runId) {

    private final CountDownLatch stopLatch = new CountDownLatch(1);

    @Override
    protected void doStart() throws Exception {
      LOG.info("Start");
    }

    @Override
    protected void doRun() throws Exception {
      stopLatch.await();
    }

    @Override
    protected void doStop() throws Exception {
      LOG.info("Stop");
    }

    @Override
    protected void triggerShutdown() {
      stopLatch.countDown();
    }
  };
}

@Test
public void testCompletion() throws ExecutionException, InterruptedException {
  Service service = new DummyService("s1", new AtomicBoolean());
  ListenableFuture<Service.State> completion = Services.getCompletionFuture(service);

  service.start();
  service.stop();

  completion.get();

  AtomicBoolean transiting = new AtomicBoolean();
  service = new DummyService("s2", transiting);
  completion = Services.getCompletionFuture(service);

  service.startAndWait();
  transiting.set(true);
  service.stop();

  try {
    completion.get();
    Assert.assertTrue(false);
  } catch (ExecutionException e) {
    // Expected
  }
}

/**
 * Attaches the given callbacks which will be invoked when the given Service enters a TERMINATED or FAILED state.
 * The callbacks are optional and may be invoked synchronously if the Service is already in one of these states.
 *
 * @param service            The Service to attach to.
 * @param terminatedCallback (Optional) A Runnable that will be invoked if the Service enters a TERMINATED state.
 * @param failureCallback    (Optional) A Runnable that will be invoked if the Service enters a FAILED state.
 * @param executor           An Executor to use for callback invocations.
 */
public static void onStop(Service service, Runnable terminatedCallback, Consumer<Throwable> failureCallback, Executor executor) {
    ShutdownListener listener = new ShutdownListener(terminatedCallback, failureCallback);
    service.addListener(listener, executor);

    // addListener() will not invoke the callbacks if the service is already in a terminal state. As such, we need to
    // manually check for these states after registering the listener and invoke the appropriate callback. The
    // ShutdownListener will make sure they are not invoked multiple times.
    Service.State state = service.state();
    if (state == Service.State.FAILED) {
        // We don't care (or know) the state from which we came, so we just pass some random one.
        listener.failed(Service.State.FAILED, service.failureCause());
    } else if (state == Service.State.TERMINATED) {
        listener.terminated(Service.State.TERMINATED);
    }
}

@Inject
public MigratorMonitor(@ScannerZooKeeper CuratorFramework curator, @SelfHostAndPort HostAndPort selfHostAndPort,
                       final ScanWorkflow workflow, final MigratorStatusDAO statusDAO,
                       final DataTools dataTools, LifeCycleRegistry lifecycle, LeaderServiceTask leaderServiceTask,
                       MetricRegistry metricRegistry) {
    super(curator, LEADER_DIR, selfHostAndPort.toString(), SERVICE_NAME, 1, TimeUnit.MINUTES,
            new Supplier<Service>() {
                @Override
                public Service get() {
                    return new LocalMigratorMonitor(workflow, statusDAO, dataTools);
                }
            });

    ServiceFailureListener.listenTo(this, metricRegistry);
    leaderServiceTask.register(SERVICE_NAME, this);
    lifecycle.manage(new ManagedGuavaService(this));
}

@Inject
public ScanUploadSchedulingService(@ScannerZooKeeper CuratorFramework curator, @SelfHostAndPort HostAndPort selfHostAndPort,
                                   final ScanUploader scanUploader, final List<ScheduledDailyScanUpload> scheduledScans,
                                   final ScanCountListener scanCountListener, final StashRequestManager stashRequestManager,
                                   LifeCycleRegistry lifecycle, LeaderServiceTask leaderServiceTask,
                                   final MetricRegistry metricRegistry,
                                   final Clock clock) {
    super(curator, LEADER_DIR, selfHostAndPort.toString(), SERVICE_NAME, 1, TimeUnit.MINUTES,
            new Supplier<Service>() {
                @Override
                public Service get() {
                    return new DelegateSchedulingService(scanUploader, stashRequestManager, scheduledScans, scanCountListener, clock);
                }
            });

    ServiceFailureListener.listenTo(this, metricRegistry);
    leaderServiceTask.register(SERVICE_NAME, this);
    lifecycle.manage(new ManagedGuavaService(this));
}

@Inject
public ScanUploadMonitor(@ScannerZooKeeper CuratorFramework curator, @SelfHostAndPort HostAndPort selfHostAndPort,
                         final ScanWorkflow scanWorkflow, final ScanStatusDAO scanStatusDAO,
                         final ScanWriterGenerator scanWriterGenerator,
                         final StashStateListener stashStateListener, final ScanCountListener scanCountListener,
                         final DataTools dataTools, LifeCycleRegistry lifecycle, LeaderServiceTask leaderServiceTask,
                         MetricRegistry metricRegistry) {
    super(curator, LEADER_DIR, selfHostAndPort.toString(), SERVICE_NAME, 1, TimeUnit.MINUTES,
            new Supplier<Service>() {
                @Override
                public Service get() {
                    return new LocalScanUploadMonitor(scanWorkflow, scanStatusDAO, scanWriterGenerator,
                            stashStateListener, scanCountListener, dataTools);
                }
            });

    ServiceFailureListener.listenTo(this, metricRegistry);
    leaderServiceTask.register(SERVICE_NAME, this);
    lifecycle.manage(new ManagedGuavaService(this));
}

private Optional<LeaderService> startService(final String name) {
    if (!isOwner(name)) {
        return Optional.absent();
    }

    _log.info("Starting owned service {}: {}", _group, name);

    String zkLeaderPath = String.format("/leader/%s/%s", _group.toLowerCase(), name);
    String threadName = String.format("Leader-%s-%s", _group, name);
    String taskName = String.format("%s-%s", _group.toLowerCase(), name);

    LeaderService leaderService = new LeaderService(_curator, zkLeaderPath, _selfId,
            threadName, 1, TimeUnit.MINUTES, new Supplier<Service>() {
        @Override
        public Service get() {
            return _factory.create(name);
        }
    });
    ServiceFailureListener.listenTo(leaderService, _metricRegistry);
    _dropwizardTask.register(taskName, leaderService);
    leaderService.start();
    return Optional.of(leaderService);
}

/** Returns true if the Guava service entered the RUNNING state within the specified time period. */
private boolean awaitRunning(Service service, long timeoutAt) {
    if (service.isRunning()) {
        return true;
    }
    long waitMillis = timeoutAt - System.currentTimeMillis();
    if (waitMillis <= 0) {
        return false;
    }
    try {
        service.start().get(waitMillis, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        // Fall through
    }
    return service.isRunning();
}

private Managed newInboundReplication(final DataCenter dataCenter) {
    // Create a proxy for the remote data center.
    final ReplicationSource replicationSource = newRemoteReplicationSource(dataCenter);

    // Start asynchronously downloading events from the remote data center.
    final Managed fanout = new GuavaServiceController(_replicationEnabled, new Supplier<Service>() {
        @Override
        public Service get() {
            return _fanoutManager.newInboundReplicationFanout(dataCenter, replicationSource);
        }
    });

    // Note: closing the replication source could also be done via a listener on the Guava service...
    return new Managed() {
        @Override
        public void start() throws Exception {
            fanout.start();
        }

        @Override
        public void stop() throws Exception {
            fanout.stop();
            ServicePoolProxies.close(replicationSource);
        }
    };
}

@Inject
SystemQueueMonitorManager(LifeCycleRegistry lifeCycle,
                          final DatabusEventStore eventStore,
                          final DataCenters dataCenters,
                          @DatabusClusterInfo final Collection<ClusterInfo> clusterInfo,
                          @DatabusZooKeeper CuratorFramework curator,
                          @SelfHostAndPort HostAndPort self,
                          LeaderServiceTask dropwizardTask,
                          final MetricRegistry metricRegistry) {
    LeaderService leaderService = new LeaderService(
            curator, "/leader/queue-monitor", self.toString(), "Leader-QueueMonitor", 1, TimeUnit.MINUTES,
            new Supplier<Service>() {
                @Override
                public Service get() {
                    return new SystemQueueMonitor(eventStore, dataCenters, clusterInfo, metricRegistry);
                }
            });
    ServiceFailureListener.listenTo(leaderService, metricRegistry);
    dropwizardTask.register("queue-monitor", leaderService);
    lifeCycle.manage(new ManagedGuavaService(leaderService));
}

/**
 * Tests the behavior of AbstractThreadPoolService when the runFuture completes (normally or not).
 */
@Test
public void testAutoShutdown() {
    // When completed normally.
    @Cleanup
    val s1 = newService();
    s1.runFuture.complete(null);
    s1.awaitTerminated();
    Assert.assertEquals("Unexpected state upon auto-shutdown (normal completion).", Service.State.TERMINATED, s1.state());

    // When completed with failure.
    @Cleanup
    val s2 = newService();
    s2.runFuture.completeExceptionally(new IntentionalException());
    AssertExtensions.assertThrows(
            "Service did not fail when runFuture failed.",
            () -> s2.awaitTerminated(),
            ex -> ex instanceof IllegalStateException);
    Assert.assertEquals("Unexpected state upon auto-shutdown (failure).", Service.State.FAILED, s2.state());
    Assert.assertTrue("Unexpected failure cause.", s2.failureCause() instanceof IntentionalException);
}

@Override
public void terminated(Service.State from) {
    log.info("{} client stopped", type.getName());
    broadcastOnDisconnect();
    failedAddresses.add(lastServerAddress);
    lastServerAddress = null;
    stratumClient = null;
    // Try to restart
    if (!stopped) {
        log.info("Reconnecting {} in {} seconds", type.getName(), retrySeconds);
        connectionExec.remove(connectionCheckTask);
        connectionExec.remove(reconnectTask);
        if (retrySeconds > 0) {
            reconnectAt = System.currentTimeMillis() + retrySeconds * 1000;
            connectionExec.schedule(reconnectTask, retrySeconds, TimeUnit.SECONDS);
        } else {
            connectionExec.execute(reconnectTask);
        }
    }
}

public void startAsync() {
    if (stratumClient == null){
        log.info("Forcing service start");
        connectionExec.remove(reconnectTask);
        createStratumClient();
    }

    Service.State state = stratumClient.state();
    if (state != NEW || stopped) {
        log.debug("Not starting service as it is already started or explicitly stopped");
        return;
    }

    try {
        stratumClient.startAsync();
    } catch (IllegalStateException e) {
        // This can happen if the service has already been started or stopped (e.g. by another
        // service or listener). Our contract says it is safe to call this method if
        // all services were NEW when it was called, and this has already been verified above, so we
        // don't propagate the exception.
        log.warn("Unable to start Service " + type.getName(), e);
    }
}

@Override
public void terminated(Service.State from) {
    log.info("{} client stopped", type.getName());
    broadcastOnDisconnect();
    failedAddresses.add(lastServerAddress);
    lastServerAddress = null;
    // Try to restart
    if (!stopped) {
        log.info("Reconnecting {} in {} seconds", type.getName(), retrySeconds);
        connectionExec.remove(reconnectTask);
        if (retrySeconds > 0) {
            connectionExec.schedule(reconnectTask, retrySeconds, TimeUnit.SECONDS);
        } else {
            connectionExec.execute(reconnectTask);
        }
    }
}

/**
 * Generates a list of services from a valid class list
 *
 * @param classes a list of classes
 * @return the instantiated services from the given class list
 */
private ServiceMap getServicesFromClassList(List<Class> classes) {
    final ServiceMap services = new ServiceMap();
    for (Class<?> klazz : classes) {
        try {
            if (klazz.isAssignableFrom(Service.class)) {
                Service service = (Service) klazz.newInstance();
                String serviceName = klazz.getAnnotation(RobotService.class).value();
                services.put(serviceName.equals("") ? klazz.getSimpleName() : serviceName, service);

                if (klazz.isAssignableFrom(LatebindingContextService.class)) {
                    robotContextServices.add(LatebindingContextService.class.cast(service));
                }
            } else {
                Log.w(TAG, klazz.getSimpleName() + " is not a valid service, it must have a super class of Service");
            }
        } catch (InstantiationException | IllegalAccessException e) {
            e.printStackTrace();
        }
    }
    return services;
}

private void shutdownWhenStopped(Service component, String componentName) {
    Consumer<Throwable> failedHandler = cause -> {
        log.warn("{}: {} failed. Shutting down StreamSegmentContainer.", this.traceObjectId, componentName, cause);
        if (state() == State.RUNNING) {
            // We can only stop the service if it's already running. During the stop it will pick up the failure cause
            // and terminate in failure.
            stopAsync();
        } else if (state() == State.STARTING) {
            // We can only notify failed if we are starting. We cannot fail a service if it's already in a terminal state.
            notifyFailed(cause);
        }
    };
    Runnable stoppedHandler = () -> {
        if (state() == State.STARTING || state() == State.RUNNING) {
            // The Component stopped but we are not in a stopping/terminal phase. We need to shut down right away.
            log.warn("{}: {} stopped unexpectedly (no error) but StreamSegmentContainer was not currently stopping. Shutting down StreamSegmentContainer.",
                    this.traceObjectId,
                    componentName);
            stopAsync();
        }
    };
    Services.onStop(component, stoppedHandler, failedHandler, this.executor);
}

@Override
public ServiceManager get() {
    final ImmutableSet<Service> allServices = ImmutableSet.<Service>builder()
            .addAll(services)
            .addAll(configuration.getServices())
            .build();

    return new ServiceManager(allServices);
}

@Test
public void testObserverCallbacks() throws Exception {
    final Path path = temporaryFolder.newFile().toPath();

    final FileObserver fileObserverSpy = spy(fileObserver);
    final NumberSuffixStrategy namingStrategy = new NumberSuffixStrategy(path);
    final PathSet pathSet = new SinglePathSet(path.toString());

    final FileReaderService readerService = new FileReaderService(
            pathSet,
            Charsets.UTF_8,
            FileInput.InitialReadPosition.START,
            mockFileInput(),
            null,
            new NewlineChunkSplitter(),
            new CollectingBuffer(),
            1024,
            250L,
            fileObserverSpy);

    readerService.startAsync();
    readerService.awaitRunning(1, TimeUnit.MINUTES);

    assertEquals("service should be running", Service.State.RUNNING, readerService.state());

    verify(fileObserverSpy).observePathSet(eq(pathSet), any(FileObserver.Listener.class));

    readerService.stopAsync();
    readerService.awaitTerminated(1, TimeUnit.MINUTES);
}

public void restoreClicked(ActionEvent event) {
    // Don't allow a restore unless this wallet is presently empty. We don't want to end up with two wallets, too
    // much complexity, even though WalletAppKit will keep the current one as a backup file in case of disaster.
    if (Main.bitcoin.wallet().getBalance().getValue() > 0) {
        informationalAlert("Wallet is not empty",
                "You must empty this wallet out before attempting to restore an older one, as mixing wallets " +
                        "together can lead to invalidated backups.");
        return;
    }

    if (aesKey != null) {
        // This is weak. We should encrypt the new seed here.
        informationalAlert("Wallet is encrypted",
                "After restore, the wallet will no longer be encrypted and you must set a new password.");
    }

    log.info("Attempting wallet restore using seed '{}' from date {}", wordsArea.getText(), datePicker.getValue());
    informationalAlert("Wallet restore in progress",
            "Your wallet will now be resynced from the Bitcoin network. This can take a long time for old wallets.");
    overlayUI.done();
    Main.instance.controller.restoreFromSeedAnimation();

    long birthday = datePicker.getValue().atStartOfDay().toEpochSecond(ZoneOffset.UTC);
    DeterministicSeed seed = new DeterministicSeed(Splitter.on(' ').splitToList(wordsArea.getText()), null, "", birthday);
    // Shut down creativecoinj and restart it with the new seed.
    Main.bitcoin.addListener(new Service.Listener() {
        @Override
        public void terminated(Service.State from) {
            Main.instance.setupWalletKit(seed);
            Main.bitcoin.startAsync();
        }
    }, Platform::runLater);
    Main.bitcoin.stopAsync();
}

public void restoreClicked(ActionEvent event) {
    // Don't allow a restore unless this wallet is presently empty. We don't want to end up with two wallets, too
    // much complexity, even though WalletAppKit will keep the current one as a backup file in case of disaster.
    if (Main.bitcoin.wallet().getBalance().value > 0) {
        informationalAlert("Wallet is not empty",
                "You must empty this wallet out before attempting to restore an older one, as mixing wallets " +
                        "together can lead to invalidated backups.");
        return;
    }

    if (aesKey != null) {
        // This is weak. We should encrypt the new seed here.
        informationalAlert("Wallet is encrypted",
                "After restore, the wallet will no longer be encrypted and you must set a new password.");
    }

    log.info("Attempting wallet restore using seed '{}' from date {}", wordsArea.getText(), datePicker.getValue());
    informationalAlert("Wallet restore in progress",
            "Your wallet will now be resynced from the Bitcoin network. This can take a long time for old wallets.");
    overlayUI.done();
    Main.instance.controller.restoreFromSeedAnimation();

    long birthday = datePicker.getValue().atStartOfDay().toEpochSecond(ZoneOffset.UTC);
    DeterministicSeed seed = new DeterministicSeed(Splitter.on(' ').splitToList(wordsArea.getText()), null, "", birthday);
    // Shut down bitcoinj and restart it with the new seed.
    Main.bitcoin.addListener(new Service.Listener() {
        @Override
        public void terminated(Service.State from) {
            Main.instance.setupWalletKit(seed);
            Main.bitcoin.startAsync();
        }
    }, Platform::runLater);
    Main.bitcoin.stopAsync();
}

private void uninitialize() {
  LOG.debug("Source exiting " + this.peerId);
  metrics.clear();
  if (replicationEndpoint.state() == Service.State.STARTING
      || replicationEndpoint.state() == Service.State.RUNNING) {
    replicationEndpoint.stopAndWait();
  }
}

@Test
public void testThriftServerStop() throws Exception {
  Assert.assertEquals(Service.State.RUNNING, txService.thriftRPCServerState());

  int nThreads = NUM_CLIENTS;
  ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
  for (int i = 0; i < nThreads; ++i) {
    executorService.submit(new Runnable() {
      @Override
      public void run() {
        try {
          TransactionSystemClient txClient = getClient();
          clientsDoneLatch.countDown();
          // this will hang, due to the slow edit log (until the latch in it is stopped)
          txClient.startShort();
        } catch (Exception e) {
          // Exception expected
        }
      }
    });
  }

  // Wait till all clients finish sending request to transaction manager
  clientsDoneLatch.await();
  TimeUnit.SECONDS.sleep(1);

  // Expire zookeeper session, which causes Thrift server to stop.
  expireZkSession(zkClientService);
  waitForThriftStop();

  // Stop Zookeeper client so that it does not re-connect to Zookeeper and start Thrift server again.
  zkClientService.stopAndWait();
  storageWaitLatch.countDown();
  TimeUnit.SECONDS.sleep(1);

  // Make sure Thrift server stopped.
  Assert.assertEquals(Service.State.TERMINATED, txService.thriftRPCServerState());
}

/**
 * Create a new {@link ServiceManagerIface} that wraps a {@link ServiceManager}.
 *
 * @param delegate Service manager to delegate to.
 * @return A wrapper.
 */
public static ServiceManagerIface serviceManager(final ServiceManager delegate) {
  return new ServiceManagerIface() {
    @Override
    public ServiceManagerIface startAsync() {
      delegate.startAsync();
      return this;
    }

    @Override
    public void awaitHealthy() {
      delegate.awaitHealthy();
    }

    @Override
    public ServiceManagerIface stopAsync() {
      delegate.stopAsync();
      return this;
    }

    @Override
    public void awaitStopped(long timeout, TimeUnit unit) throws TimeoutException {
      delegate.awaitStopped(timeout, unit);
    }

    @Override
    public ImmutableMultimap<State, Service> servicesByState() {
      return delegate.servicesByState();
    }
  };
}

@Provides
@Singleton
@AppStartup
ServiceManagerIface provideAppStartupServiceManager(
    @AppStartup Set<Service> services,
    LifecycleShutdownListener listener) {

  ServiceManager manager = new ServiceManager(services);
  manager.addListener(listener);
  return GuavaUtils.serviceManager(manager);
}

@Provides
@Singleton
@SchedulerActive
ServiceManagerIface provideSchedulerActiveServiceManager(
    @SchedulerActive Set<Service> services,
    LifecycleShutdownListener listener) {

  ServiceManager manager = new ServiceManager(services);
  manager.addListener(listener);
  return GuavaUtils.serviceManager(manager);
}

/**
 * Starts the pubsub system and gets a handle to the event sink where pubsub events may be sent.
 *
 * @param injector Injector where the pubsub system was installed.
 * @return The pubsub event sink.
 * @throws Exception If the pubsub system failed to start.
 */
public static EventSink startPubsub(Injector injector) throws Exception {
  // TODO(wfarner): Make it easier to write a unit test wired for pubsub events.
  // In this case, a trade-off was made to avoid installing several distant modules and providing
  // required bindings that seem unrelated from this code.
  Set<Service> services = injector.getInstance(
      Key.get(new TypeLiteral<Set<Service>>() { }, AppStartup.class));

  for (Service service : services) {
    service.startAsync().awaitRunning();
  }
  return injector.getInstance(EventSink.class);
}

@Test
public void testGetServices() throws Exception {
  Service newService = createMock(Service.class);
  expect(newService.state()).andReturn(State.NEW);

  Service failedService = createMock(Service.class);
  expect(failedService.state()).andReturn(State.FAILED);
  Exception failureCause = new Exception(FAILURE_CAUSE_REASON);
  expect(failedService.failureCause()).andReturn(failureCause);

  Service runningService = createMock(Service.class);
  expect(runningService.state()).andReturn(State.RUNNING);

  expect(startupServiceManager.servicesByState()).andReturn(
      ImmutableMultimap.of(
          State.RUNNING, runningService,
          State.FAILED, failedService));
  expect(activeServicesManager.servicesByState())
      .andReturn(ImmutableMultimap.of(State.NEW, newService));

  control.replay();

  assertEquals(
      ImmutableList.of(
          ImmutableMap.of(
              "name", newService.getClass().getSimpleName(),
              "state", State.NEW),
          ImmutableMap.of(
              "name", failedService.getClass().getSimpleName(),
              "state", State.RUNNING),
          ImmutableMap.of(
              "name", failedService.getClass().getSimpleName(),
              "state", State.FAILED,
              "failureCause", failureCause.toString())),
      servicesServlet.getServices().getEntity());
}

/**
 * Tests the behavior of AbstractThreadPoolService during shutdown when only a RunException is present.
 */
@Test
public void testShutdownRunException() {
    @Cleanup
    val s = newService();

    s.stopAsync();
    s.runFuture.completeExceptionally(new IntentionalException());
    AssertExtensions.assertThrows(
            "Service did not fail when runFuture failed.",
            () -> s.awaitTerminated(),
            ex -> ex instanceof IllegalStateException);
    Assert.assertEquals("Unexpected state upon failed shutdown.", Service.State.FAILED, s.state());
    Assert.assertTrue("Unexpected failure cause.", s.failureCause() instanceof IntentionalException);
}