org.junit.jupiter.api.AfterAll#com.google.common.util.concurrent.MoreExecutors源码实例Demo

@Override
public AgentConnection register(String agentId, int agentVersion, Channel channel) {
    DefaultAgentConnection agentConnection = new DefaultAgentConnection(agentId, agentVersion, channel);
    AgentConnection oldConnection = connections.get(agentId);
    if (!Objects.equals(oldConnection, agentConnection)) {
        oldConnection = connections.put(agentId, agentConnection);
        agentConnection.init();
        agentConnection.closeFuture().addListener(() -> connections.remove(agentId, agentConnection), MoreExecutors.directExecutor());
        if (oldConnection != null && !Objects.equals(oldConnection, agentConnection)) {
            oldConnection.close();
        }
        return agentConnection;
    } else {
        return oldConnection;
    }
}
 

public void testBulkLoadInterruptedException() {
  Exception e = new InterruptedException();
  CacheLoader<Object, Object> loader = exceptionLoader(e);
  LoadingCache<Object, Object> cache = CaffeinatedGuava.build(Caffeine.newBuilder()
      .recordStats().executor(MoreExecutors.directExecutor()), bulkLoader(loader));
  CacheStats stats = cache.stats();
  assertEquals(0, stats.missCount());
  assertEquals(0, stats.loadSuccessCount());
  assertEquals(0, stats.loadExceptionCount());
  assertEquals(0, stats.hitCount());

  try {
    cache.getAll(asList(new Object()));
    fail();
  } catch (ExecutionException expected) {
    assertSame(e, expected.getCause());
  }
  assertTrue(Thread.interrupted());
  stats = cache.stats();
  assertEquals(1, stats.missCount());
  assertEquals(0, stats.loadSuccessCount());
  assertEquals(1, stats.loadExceptionCount());
  assertEquals(0, stats.hitCount());
}
 

@Before
public void setUp() throws Exception {
  //noinspection unchecked
  graphBuilder = mock(GraphBuilder.class);
  when(graphBuilder.getFallback())
      .thenReturn(Optional.<AsyncFunction<Throwable, String>>absent());

  Map<Input<?>, Object> emptyMap = Collections.emptyMap();
  traverseState = new TraverseState(emptyMap, MoreExecutors.sameThreadExecutor(), true);

  List<? extends NodeInfo> currentNodeParameters = ImmutableList.of();

  currentNodeInfo = new FakeNodeInfo("the node", currentNodeParameters);
  List<ListenableFuture<?>> currentNodeValues = ImmutableList.of();

  currentCall = new TraverseState.FutureCallInformation(currentNodeInfo, currentNodeValues);
  currentCallInfo = new CallInfo(currentNodeInfo, NO_PARAMS);

  fallback = new NodeExecutionFallback<String>(graphBuilder, currentCall, traverseState);
}
 

/**
 * Register a callback to cache the result of a check.
 * @param target
 * @param lf
 */
private void addResultCachingCallback(
    Checkable<K, V> target, ListenableFuture<V> lf) {
  Futures.addCallback(lf, new FutureCallback<V>() {
    @Override
    public void onSuccess(@Nullable V result) {
      synchronized (ThrottledAsyncChecker.this) {
        checksInProgress.remove(target);
        completedChecks.put(target, new LastCheckResult<>(
            result, timer.monotonicNow()));
      }
    }

    @Override
    public void onFailure(@Nonnull Throwable t) {
      synchronized (ThrottledAsyncChecker.this) {
        checksInProgress.remove(target);
        completedChecks.put(target, new LastCheckResult<>(
            t, timer.monotonicNow()));
      }
    }
  }, MoreExecutors.directExecutor());
}
 

ListenableFuture<Void> close() {
  if (isClosed.getAndSet(true)) {
    return closeFuture;
  } else if (closeFuture.isDone()) {
    return closeFuture;
  }

  // Close abruptly if the closing future is cancelled.
  closeFuture.addListener(
      () -> {
        if (closeFuture.isCancelled()) {
          closeNow();
        }
      },
      MoreExecutors.directExecutor());

  try {
    pendingWrites.put(CLOSE_EVENT_FUTURE);
  } catch (InterruptedException e) {
    closeNow();
    logger.atSevere().withCause(e).log("Failed to close the sequential writer.");
    closeFuture.set(null);
  }
  return closeFuture;
}
 

/**
 * Executes a query to retrieve all the table rows
 *
 * @param callback Callback to invoke when the operation is completed
 * @throws com.microsoft.windowsazure.mobileservices.MobileServiceException
 * @deprecated use {@link #execute()} instead
 */
public void execute(final TableQueryCallback<E> callback) throws MobileServiceException {

    ListenableFuture<MobileServiceList<E>> executeFuture = execute();

    Futures.addCallback(executeFuture, new FutureCallback<MobileServiceList<E>>() {
        @Override
        public void onFailure(Throwable exception) {
            if (exception instanceof Exception) {
                callback.onCompleted(null, 0, (Exception) exception, MobileServiceException.getServiceResponse(exception));
            } else {
                callback.onCompleted(null, 0, new Exception(exception), MobileServiceException.getServiceResponse(exception));
            }
        }

        @Override
        public void onSuccess(MobileServiceList<E> result) {
            callback.onCompleted(result, result.getTotalCount(), null, null);
        }
    }, MoreExecutors.directExecutor());
}
 

@Override
public CompletableFuture<InstanceCommunication.MetricsData> getMetrics(int instanceId) {
    CompletableFuture<InstanceCommunication.MetricsData> retval = new CompletableFuture<>();
    if (stub == null) {
        retval.completeExceptionally(new RuntimeException("Not alive"));
        return retval;
    }
    ListenableFuture<InstanceCommunication.MetricsData> response = stub.withDeadlineAfter(GRPC_TIMEOUT_SECS, TimeUnit.SECONDS).getMetrics(Empty.newBuilder().build());
    Futures.addCallback(response, new FutureCallback<InstanceCommunication.MetricsData>() {
        @Override
        public void onFailure(Throwable throwable) {
            retval.completeExceptionally(throwable);
        }

        @Override
        public void onSuccess(InstanceCommunication.MetricsData t) {
            retval.complete(t);
        }
    }, MoreExecutors.directExecutor());
    return retval;
}
 

@Test
default void testBackupAndRestoreH2() throws Exception {
    File tempFile = Files.createTempFile(system().getPlanFiles().getDataFolder().toPath(), "backup-", ".db").toFile();
    tempFile.deleteOnExit();
    H2DB backup = system().getDatabaseSystem().getH2Factory().usingFile(tempFile);
    backup.setTransactionExecutorServiceProvider(MoreExecutors::newDirectExecutorService);
    try {
        backup.init();

        saveDataForBackup();

        backup.executeTransaction(new BackupCopyTransaction(db(), backup));

        assertQueryResultIsEqual(db(), backup, BaseUserQueries.fetchAllBaseUsers());
        assertQueryResultIsEqual(db(), backup, UserInfoQueries.fetchAllUserInformation());
        assertQueryResultIsEqual(db(), backup, NicknameQueries.fetchAllNicknameData());
        assertQueryResultIsEqual(db(), backup, GeoInfoQueries.fetchAllGeoInformation());
        assertQueryResultIsEqual(db(), backup, SessionQueries.fetchAllSessions());
        assertQueryResultIsEqual(db(), backup, LargeFetchQueries.fetchAllWorldNames());
        assertQueryResultIsEqual(db(), backup, LargeFetchQueries.fetchAllTPSData());
        assertQueryResultIsEqual(db(), backup, ServerQueries.fetchPlanServerInformation());
        assertQueryResultIsEqual(db(), backup, WebUserQueries.fetchAllUsers());
    } finally {
        backup.close();
    }
}
 

ListenableFuture<Void> setLocalDescription(final SessionDescription sessionDescription) {
    Log.d(EXTENDED_LOGGING_TAG, "setting local description:");
    for (final String line : sessionDescription.description.split(eu.siacs.conversations.xmpp.jingle.SessionDescription.LINE_DIVIDER)) {
        Log.d(EXTENDED_LOGGING_TAG, line);
    }
    return Futures.transformAsync(getPeerConnectionFuture(), peerConnection -> {
        final SettableFuture<Void> future = SettableFuture.create();
        peerConnection.setLocalDescription(new SetSdpObserver() {
            @Override
            public void onSetSuccess() {
                future.set(null);
            }

            @Override
            public void onSetFailure(final String s) {
                future.setException(new IllegalArgumentException("unable to set local session description: " + s));

            }
        }, sessionDescription);
        return future;
    }, MoreExecutors.directExecutor());
}
 

private synchronized void addRoutes(final MpReachNlri nlri, final org.opendaylight.yang.gen.v1.urn.opendaylight
        .params.xml.ns.yang.bgp.message.rev200120.path.attributes.Attributes attributes) {
    final TablesKey key = new TablesKey(nlri.getAfi(), nlri.getSafi());
    final TableContext ctx = this.tables.get(key);

    if (ctx == null) {
        LOG.debug("No table for {}, not accepting NLRI {}", key, nlri);
        return;
    }

    final DOMDataTreeWriteTransaction tx = this.chain.newWriteOnlyTransaction();
    ctx.writeRoutes(tx, nlri, attributes);
    LOG.trace("Write routes {}", nlri);
    tx.commit().addCallback(new FutureCallback<CommitInfo>() {
        @Override
        public void onSuccess(final CommitInfo result) {
            LOG.trace("Successful commit");
        }

        @Override
        public void onFailure(final Throwable trw) {
            LOG.error("Failed commit", trw);
        }
    }, MoreExecutors.directExecutor());
}
 

public void testGet() {
  Object computed = new Object();
  LoadingCache<Object, Object> cache = CaffeinatedGuava.build(Caffeine.newBuilder()
      .executor(MoreExecutors.directExecutor())
      .maximumSize(0)
      .removalListener(listener),
      constantLoader(computed));

  Object key = new Object();
  assertSame(computed, cache.getUnchecked(key));
  RemovalNotification<Object, Object> notification = listener.remove();
  assertSame(key, notification.getKey());
  assertSame(computed, notification.getValue());
  assertSame(RemovalCause.SIZE, notification.getCause());
  assertTrue(listener.isEmpty());
  checkEmpty(cache);
}
 

@Test
public void advertisedEncodingsAreSent() {
  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      MoreExecutors.directExecutor(),
      baseCallOptions,
      provider,
      deadlineCancellationExecutor,
      channelCallTracer,
      false /* retryEnabled */)
          .setDecompressorRegistry(decompressorRegistry);

  call.start(callListener, new Metadata());

  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
  verify(transport).newStream(eq(method), metadataCaptor.capture(), same(baseCallOptions));
  Metadata actual = metadataCaptor.getValue();

  // there should only be one.
  Set<String> acceptedEncodings = ImmutableSet.of(
      new String(actual.get(GrpcUtil.MESSAGE_ACCEPT_ENCODING_KEY), GrpcUtil.US_ASCII));
  assertEquals(decompressorRegistry.getAdvertisedMessageEncodings(), acceptedEncodings);
}
 

@Test
public void cancel() throws Exception {
    when(armeriaCall.tryFinish()).thenReturn(false);
    when(armeriaCall.isCanceled()).thenReturn(false, false, true);

    final ManualMockCallback callback = new ManualMockCallback();
    final StreamingCallSubscriber subscriber = new StreamingCallSubscriber(
            armeriaCall, callback, new Request.Builder().url("http://foo.com").build(),
            MoreExecutors.directExecutor());
    subscriber.onSubscribe(subscription);
    subscriber.onNext(ResponseHeaders.of(200));
    subscriber.onNext(HttpData.ofUtf8("{\"name\":\"foo\"}"));
    subscriber.onComplete();

    verify(subscription, times(2)).request(1L);

    await().untilAsserted(() -> assertThat(callback.callbackCallingCount).isEqualTo(1));
    await().untilAsserted(() -> assertThat(callback.exception.getMessage()).isEqualTo("cancelled"));
}
 

@Override
protected void configure(ServerBuilder sb) throws Exception {
    sb.service("/sse/publisher", (ctx, req) -> ServerSentEvents.fromPublisher(
            Flux.just(ServerSentEvent.ofData("foo"), ServerSentEvent.ofData("bar"))));
    sb.service("/sse/stream", (ctx, req) -> ServerSentEvents.fromStream(
            Stream.of(ServerSentEvent.ofData("foo"), ServerSentEvent.ofData("bar")),
            MoreExecutors.directExecutor()));

    sb.service("/converter/publisher", (ctx, req) -> ServerSentEvents.fromPublisher(
            Flux.just("foo", "bar"), ServerSentEvent::ofComment));
    sb.service("/converter/stream", (ctx, req) -> ServerSentEvents.fromStream(
            Stream.of("foo", "bar"), MoreExecutors.directExecutor(), ServerSentEvent::ofComment));

    sb.service("/single/sse", (ctx, req) -> ServerSentEvents.fromEvent(
            ServerSentEvent.ofEvent("add")));
}
 

@Activate
public void activate(ComponentContext context) {
    configService.registerProperties(getClass());
    setupCodecs();
    messageReceivingExecutor = receiveOnIOLoopThread
            ? MoreExecutors.directExecutor()
            : Executors.newFixedThreadPool(
                    totalReceiverThreads,
                    groupedThreads("onos/net-perf-test", "receiver-%d"));
    registerMessageHandlers();
    startTest();
    reporter.scheduleWithFixedDelay(this::reportPerformance,
            reportIntervalSeconds,
            reportIntervalSeconds,
            TimeUnit.SECONDS);
    logConfig("Started");
}
 

/**
 * Invokes a custom API
 *
 * @param apiName        The API name
 * @param content        The byte array to send as the request body
 * @param httpMethod     The HTTP Method used to invoke the API
 * @param requestHeaders The extra headers to send in the request
 * @param parameters     The query string parameters sent in the request
 * @param callback       The callback to invoke after the API execution
 */
public void invokeApi(String apiName, byte[] content, String httpMethod, List<Pair<String, String>> requestHeaders, List<Pair<String, String>> parameters,
                      final ServiceFilterResponseCallback callback) {

    ListenableFuture<ServiceFilterResponse> invokeApiFuture = invokeApi(apiName, content, httpMethod, requestHeaders, parameters);

    Futures.addCallback(invokeApiFuture, new FutureCallback<ServiceFilterResponse>() {
        @Override
        public void onFailure(Throwable exception) {
            if (exception instanceof Exception) {
                callback.onResponse(MobileServiceException.getServiceResponse(exception), (Exception) exception);
            } else {
                callback.onResponse(MobileServiceException.getServiceResponse(exception), new Exception(exception));
            }
        }

        @Override
        public void onSuccess(ServiceFilterResponse result) {
            callback.onResponse(result, null);
        }
    }, MoreExecutors.directExecutor());
}
 

/**
 * Build a mocked bitfinex connection
 * @return
 */
public static BitfinexWebsocketClient buildMockedBitfinexConnection() {

	final ExecutorService executorService = MoreExecutors.newDirectExecutorService();
	final BitfinexWebsocketClient bitfinexApiBroker = Mockito.mock(SimpleBitfinexApiBroker.class);
	final BitfinexWebsocketConfiguration config = Mockito.mock(BitfinexWebsocketConfiguration.class);

	Mockito.when(bitfinexApiBroker.getConfiguration()).thenReturn(config);
	Mockito.when(config.getApiKey()).thenReturn(API_KEY);
	Mockito.when(bitfinexApiBroker.isAuthenticated()).thenReturn(true);
	Mockito.when(bitfinexApiBroker.getApiKeyPermissions()).thenReturn(BitfinexApiKeyPermissions.ALL_PERMISSIONS);
	Mockito.when(bitfinexApiBroker.getCallbacks()).thenReturn(new BitfinexApiCallbackRegistry());

	final OrderManager orderManager = new OrderManager(bitfinexApiBroker, executorService);
	final TradeManager tradeManager = new TradeManager(bitfinexApiBroker, executorService);
	Mockito.when(bitfinexApiBroker.getOrderManager()).thenReturn(orderManager);
	Mockito.when(bitfinexApiBroker.getTradeManager()).thenReturn(tradeManager);

	return bitfinexApiBroker;
}
 

/**
 * Inserts a JsonObject into a Mobile Service Table
 *
 * @param element    The JsonObject to insert
 * @param parameters A list of user-defined parameters and values to include in the
 *                   request URI query string
 * @param callback   Callback to invoke when the operation is completed
 * @throws IllegalArgumentException if the element has an id property set with a numeric value
 *                                  other than default (0), or an invalid string value
 * @deprecated use {@link #insert(JsonObject element, List parameters)} instead
 */
public void insert(final JsonObject element, List<Pair<String, String>> parameters, final TableJsonOperationCallback callback) {
    ListenableFuture<JsonObject> insertFuture = insert(element, parameters);

    Futures.addCallback(insertFuture, new FutureCallback<JsonObject>() {
        @Override
        public void onFailure(Throwable exception) {
            if (exception instanceof Exception) {
                callback.onCompleted(null, (Exception) exception, MobileServiceException.getServiceResponse(exception));
            } else {
                callback.onCompleted(null, new Exception(exception), MobileServiceException.getServiceResponse(exception));
            }
        }

        @Override
        public void onSuccess(JsonObject result) {
            callback.onCompleted(result, null, null);
        }
    }, MoreExecutors.directExecutor());
}
 

/**
 * Registers the client for push notification using device {@link Installation}
 *
 * @param installation device installation in Azure Notification Hub (https://msdn.microsoft.com/en-us/library/azure/mt621153.aspx)
 * @return Future with registration information
 */
public ListenableFuture<Void> register(Installation installation) {
    final SettableFuture<Void> resultFuture = SettableFuture.create();

    ListenableFuture<Void> registerInternalFuture = createOrUpdateInstallation(installation);

    Futures.addCallback(registerInternalFuture, new FutureCallback<Void>() {
        @Override
        public void onFailure(Throwable exception) {
            resultFuture.setException(exception);
        }

        @Override
        public void onSuccess(Void v) {
            resultFuture.set(v);
        }
    }, MoreExecutors.directExecutor());

    return resultFuture;
}
 

public void testAsMapKeySet_contains() {
  Cache<Integer, Integer> cache = CaffeinatedGuava.build(Caffeine.newBuilder()
      .expireAfterWrite(1000, TimeUnit.MILLISECONDS)
      .executor(MoreExecutors.directExecutor())
      .ticker(fakeTicker::read));

  cache.put(10, 20);
  fakeTicker.advance(500, TimeUnit.MILLISECONDS);
  cache.put(20, 22);
  cache.put(5, 10);

  fakeTicker.advance(501, TimeUnit.MILLISECONDS);

  assertTrue(cache.asMap().keySet().contains(20));
  assertTrue(cache.asMap().keySet().contains(5));
  assertFalse(cache.asMap().keySet().contains(10));
}
 

@Override
public ListenableFuture<Void> publish(
    int clientId, int sequenceNumber, long publishTimestampMillis) {
  SettableFuture<Void> done = SettableFuture.create();
  ApiFutures.addCallback(
      publisher.publish(
          PubsubMessage.newBuilder()
              .setData(payload)
              .putAttributes("sendTime", Long.toString(publishTimestampMillis))
              .putAttributes("clientId", Integer.toString(clientId))
              .putAttributes("sequenceNumber", Integer.toString(sequenceNumber))
              .build()),
      new ApiFutureCallback<String>() {
        @Override
        public void onSuccess(String messageId) {
          done.set(null);
        }

        @Override
        public void onFailure(Throwable t) {
          done.setException(t);
        }
      },
      MoreExecutors.directExecutor());
  return done;
}
 

public static ApiFuture<SearchGoogleAdsFieldsPagedResponse> createAsync(
    PageContext<SearchGoogleAdsFieldsRequest, SearchGoogleAdsFieldsResponse, GoogleAdsField>
        context,
    ApiFuture<SearchGoogleAdsFieldsResponse> futureResponse) {
  ApiFuture<SearchGoogleAdsFieldsPage> futurePage =
      SearchGoogleAdsFieldsPage.createEmptyPage().createPageAsync(context, futureResponse);
  return ApiFutures.transform(
      futurePage,
      new ApiFunction<SearchGoogleAdsFieldsPage, SearchGoogleAdsFieldsPagedResponse>() {
        @Override
        public SearchGoogleAdsFieldsPagedResponse apply(SearchGoogleAdsFieldsPage input) {
          return new SearchGoogleAdsFieldsPagedResponse(input);
        }
      },
      MoreExecutors.directExecutor());
}
 

@Override
public void processResponseStream(StreamContext<ProcessEvent> context) {
    Consumer<ProcessEvent> watcher = context.getWatcher();
    InputStream response = context.getStream();
    SettableFuture<Boolean> interrupter = context.getInterrupter();
    interrupter.addListener(() -> Thread.currentThread().interrupt(), MoreExecutors.directExecutor());
    try (FrameReader frameReader = new FrameReader(response)) {

        Frame frame = frameReader.readFrame();
        while (frame != null && !interrupter.isDone()) {
            try {
                ProcessEvent.watchRaw(watcher, frame.getMessage(), false);
            } catch (Exception e) {
                log.error("Cannot read body", e);
            } finally {
                frame = frameReader.readFrame();
            }
        }
    } catch (Exception t) {
        log.error("Cannot close reader", t);
    }

}
 

/**
 * Invokes a custom API
 *
 * @param apiName    The API name
 * @param body       The json element to send as the request body
 * @param httpMethod The HTTP Method used to invoke the API
 * @param parameters The query string parameters sent in the request
 * @param callback   The callback to invoke after the API execution
 * @deprecated use {@link #invokeApi(String apiName, com.google.gson.JsonElement body, String httpMethod, List parameters)} instead
 */
public void invokeApi(String apiName, JsonElement body, String httpMethod, List<Pair<String, String>> parameters, final ApiJsonOperationCallback callback) {

    ListenableFuture<JsonElement> invokeApiFuture = invokeApi(apiName, body, httpMethod, parameters);

    Futures.addCallback(invokeApiFuture, new FutureCallback<JsonElement>() {
        @Override
        public void onFailure(Throwable exception) {
            if (exception instanceof Exception) {
                callback.onCompleted(null, (Exception) exception, MobileServiceException.getServiceResponse(exception));
            } else {
                callback.onCompleted(null, new Exception(exception), MobileServiceException.getServiceResponse(exception));
            }
        }

        @Override
        public void onSuccess(JsonElement result) {
            callback.onCompleted(result, null, null);
        }
    }, MoreExecutors.directExecutor());
}
 

@Test
public void completeOnlyHeaders() throws Exception {
    when(armeriaCall.tryFinish()).thenReturn(true);

    final ManualMockCallback callback = new ManualMockCallback();
    final StreamingCallSubscriber subscriber = new StreamingCallSubscriber(
            armeriaCall, callback, new Request.Builder().url("http://foo.com").build(),
            MoreExecutors.directExecutor());
    subscriber.onSubscribe(subscription);
    subscriber.onNext(ResponseHeaders.of(HttpStatus.OK, CONTENT_LENGTH, 0));
    subscriber.onComplete();

    verify(subscription, times(2)).request(1L);
    await().untilAsserted(() -> assertThat(callback.callbackCallingCount).isEqualTo(1));
    await().untilAsserted(
            () -> assertThat(callback.response.header(CONTENT_LENGTH.toString())).isEqualTo("0"));
    await().untilAsserted(() -> assertThat(callback.response.body().string()).isEmpty());
}
 

public static PerBuildState create(Parser parser, Cell cell) {
  return parser
      .getPerBuildStateFactory()
      .create(
          ParsingContext.builder(cell, MoreExecutors.newDirectExecutorService())
              .setSpeculativeParsing(SpeculativeParsing.ENABLED)
              .build(),
          parser.getPermState());
}
 

@Before
public void setUp() {
  executor = Executors.newCachedThreadPool(
      new ThreadFactoryBuilder().setNameFormat("SlowRead-%d").setDaemon(true).build());
  addTearDown(() -> MoreExecutors.shutdownAndAwaitTermination(executor, 1, TimeUnit.SECONDS));
  storage = MemStorageModule.newEmptyStorage();
}
 

public void testComputeEviction() {
  Cache<String, String> c = CaffeinatedGuava.build(
      Caffeine.newBuilder().executor(MoreExecutors.directExecutor()).maximumSize(1));

  assertThat(c.asMap().compute("hash-1", (k, v) -> "a")).isEqualTo("a");
  assertThat(c.asMap().compute("hash-1", (k, v) -> "b")).isEqualTo("b");
  assertThat(c.asMap().compute("hash-1", (k, v) -> "c")).isEqualTo("c");
  assertThat(c.size()).isEqualTo(1);
  assertThat(c.asMap().computeIfAbsent("hash-2", k -> "")).isEqualTo("");
}
 

@Before
public void setUp() throws Exception {
  storageUtil = new StorageTestUtil(this);
  storageUtil.expectOperations();
  stateManager = createMock(StateManager.class);
  slaManager = createMock(SlaManager.class);
  TaskEventBatchWorker batchWorker = createMock(TaskEventBatchWorker.class);
  expectBatchExecute(batchWorker, storageUtil.storage, control).anyTimes();

  Injector injector = Guice.createInjector(
      new PubsubEventModule(),
      new MaintenanceModule(new MaintenanceModule.Options()),
      new AbstractModule() {
        @Override
        protected void configure() {
          bind(Storage.class).toInstance(storageUtil.storage);
          bind(StateManager.class).toInstance(stateManager);
          bind(SlaManager.class).toInstance(slaManager);
          bind(StatsProvider.class).toInstance(new FakeStatsProvider());
          bind(Executor.class).annotatedWith(AsyncExecutor.class)
              .toInstance(MoreExecutors.directExecutor());
          bind(TaskEventBatchWorker.class).toInstance(batchWorker);
          bind(new TypeLiteral<Amount<Long, Time>>() { })
              .annotatedWith(
                  MaintenanceController.MaintenanceControllerImpl.PollingInterval.class)
              .toInstance(new TimeAmount(1, Time.MINUTES));
        }
      });
  maintenance = injector.getInstance(MaintenanceController.MaintenanceControllerImpl.class);
  eventSink = PubsubTestUtil.startPubsub(injector);
}
 

public static FailureCase schedule(final InstructionScheduler scheduler, final AbstractInstructionExecutor fwd) {
    final SubmitInstructionInput input = fwd.getInput();
    final ListenableFuture<Instruction> listenableFuture;
    try {
        listenableFuture = scheduler.scheduleInstruction(input);
    } catch (final SchedulerException e) {
        LOG.info("Instuction {} failed to schedule", input, e);
        return new FailureCaseBuilder().setFailure(e.getFailure()).build();
    }
    Futures.addCallback(listenableFuture, fwd, MoreExecutors.directExecutor());
    return null;
}