com.google.common.util.concurrent.MoreExecutors#directExecutor ( )源码实例Demo

@Test
public void contextDeadlineShouldBePropagatedToStream() {
  Context context = Context.current()
      .withDeadlineAfter(1000, TimeUnit.MILLISECONDS, deadlineCancellationExecutor);
  Context origContext = context.attach();

  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      MoreExecutors.directExecutor(),
      baseCallOptions,
      provider,
      deadlineCancellationExecutor,
      channelCallTracer,
      false /* retryEnabled */);
  call.start(callListener, new Metadata());

  context.detach(origContext);

  ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class);
  verify(stream).setDeadline(deadlineCaptor.capture());

  assertTimeoutBetween(deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS), 600, 1000);
}
 

@Test
public void contextDeadlineShouldOverrideLargerCallOptionsDeadline() {
  Context context = Context.current()
      .withDeadlineAfter(1000, TimeUnit.MILLISECONDS, deadlineCancellationExecutor);
  Context origContext = context.attach();

  CallOptions callOpts = baseCallOptions.withDeadlineAfter(2000, TimeUnit.MILLISECONDS);
  ClientCallImpl<Void, Void> call = new ClientCallImpl<Void, Void>(
      method,
      MoreExecutors.directExecutor(),
      callOpts,
      provider,
      deadlineCancellationExecutor,
      channelCallTracer,
      false /* retryEnabled */);
  call.start(callListener, new Metadata());

  context.detach(origContext);

  ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class);
  verify(stream).setDeadline(deadlineCaptor.capture());

  assertTimeoutBetween(deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS), 600, 1000);
}
 

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

    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\":"));
    subscriber.onError(new IOException("foo"));

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

    await().untilAsserted(() -> assertThat(callback.callbackCallingCount).isEqualTo(1));
    await().untilAsserted(() -> assertThat(callback.exception).isNull());
    assertThatThrownBy(() -> callback.response.body().string()).hasMessageEndingWith("foo");
}
 

@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"));
}
 

@Test
public void contextDeadlineShouldOverrideLargerCallOptionsDeadline() {
  Context context = Context.current()
      .withDeadlineAfter(1000, TimeUnit.MILLISECONDS, deadlineCancellationExecutor);
  Context origContext = context.attach();

  CallOptions callOpts = baseCallOptions.withDeadlineAfter(2000, TimeUnit.MILLISECONDS);
  ClientCallImpl<Void, Void> call = new ClientCallImpl<>(
      method,
      MoreExecutors.directExecutor(),
      callOpts,
      provider,
      deadlineCancellationExecutor,
      channelCallTracer,
      /* retryEnabled= */ false);
  call.start(callListener, new Metadata());

  context.detach(origContext);

  ArgumentCaptor<Deadline> deadlineCaptor = ArgumentCaptor.forClass(Deadline.class);
  verify(stream).setDeadline(deadlineCaptor.capture());

  assertTimeoutBetween(deadlineCaptor.getValue().timeRemaining(TimeUnit.MILLISECONDS), 600, 1000);
}
 

@Before
public void setUp() {
  DEFAULTS_MAP.put("first_default_key", "first_default_value");
  DEFAULTS_MAP.put("second_default_key", "second_default_value");
  DEFAULTS_MAP.put("third_default_key", "third_default_value");

  MockitoAnnotations.initMocks(this);
  Executor directExecutor = MoreExecutors.directExecutor();

  Context context = getInstrumentation().getTargetContext();
  FirebaseApp.clearInstancesForTest();
  FirebaseApp firebaseApp =
      FirebaseApp.initializeApp(
          context,
          new FirebaseOptions.Builder()
              .setApiKey(API_KEY)
              .setApplicationId(APP_ID)
              .setProjectId(PROJECT_ID)
              .build());

  // Catch all to avoid NPEs (the getters should never return null).
  when(mockFetchedCache.get()).thenReturn(Tasks.forResult(null));
  when(mockActivatedCache.get()).thenReturn(Tasks.forResult(null));
  when(mockFireperfFetchedCache.get()).thenReturn(Tasks.forResult(null));
  when(mockFireperfActivatedCache.get()).thenReturn(Tasks.forResult(null));

  frc =
      new FirebaseRemoteConfig(
          context,
          firebaseApp,
          mockFirebaseInstallations,
          mockFirebaseAbt,
          directExecutor,
          mockFetchedCache,
          mockActivatedCache,
          mockDefaultsCache,
          mockFetchHandler,
          mockGetHandler,
          metadataClient);
}
 

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

  verify(stream, never()).setDeadline(any(Deadline.class));
}
 

private static Result doSearch(ValidatePredicateSearcher searcher, String yqlQuery, String command) {
    QueryTree queryTree = new YqlParser(new ParserEnvironment()).parse(new Parsable().setQuery(yqlQuery));
    Query query = new Query();
    query.getModel().getQueryTree().setRoot(queryTree.getRoot());

    SearchDefinition searchDefinition = new SearchDefinition("document");
    Index index = new Index("predicate_field");
    index.setPredicate(true);
    index.addCommand(command);
    searchDefinition.addIndex(index);
    IndexFacts indexFacts = new IndexFacts(new IndexModel(searchDefinition));
    Execution.Context context = new Execution.Context(null, indexFacts, null, new RendererRegistry(MoreExecutors.directExecutor()), new SimpleLinguistics());
    return new Execution(searcher, context).search(query);
}
 

@Test
public void getAttributes() {
  ClientCallImpl<Void, Void> call = new ClientCallImpl<>(
      method, MoreExecutors.directExecutor(), baseCallOptions, provider,
      deadlineCancellationExecutor, channelCallTracer, /* retryEnabled= */ false);
  Attributes attrs =
      Attributes.newBuilder().set(Key.<String>create("fake key"), "fake value").build();
  when(stream.getAttributes()).thenReturn(attrs);

  assertNotEquals(attrs, call.getAttributes());

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

  assertEquals(attrs, call.getAttributes());
}
 

@Test
public void     testFlush() throws Exception
{
    final Timing                      timing = new Timing();
    final CountDownLatch              latch = new CountDownLatch(1);
    DistributedQueue<TestQueueItem>   queue = null;
    final CuratorFramework            client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
    client.start();
    try
    {
        final AtomicBoolean     firstTime = new AtomicBoolean(true);
        queue = new DistributedQueue<TestQueueItem>(client, null, serializer, "/test", new ThreadFactoryBuilder().build(), MoreExecutors.directExecutor(), 10, true, null, QueueBuilder.NOT_SET, true, 0)
        {
            @Override
            void internalCreateNode(final String path, final byte[] bytes, final BackgroundCallback callback) throws Exception
            {
                if ( firstTime.compareAndSet(true, false) )
                {
                    Executors.newSingleThreadExecutor().submit
                    (
                        new Callable<Object>()
                        {
                            @Override
                            public Object call() throws Exception
                            {
                                latch.await();
                                timing.sleepABit();
                                client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).inBackground(callback).forPath(path, bytes);
                                return null;
                            }
                        }
                    );
                }
                else
                {
                    super.internalCreateNode(path, bytes, callback);
                }
            }
        };
        queue.start();

        queue.put(new TestQueueItem("1"));
        Assert.assertFalse(queue.flushPuts(timing.forWaiting().seconds(), TimeUnit.SECONDS));
        latch.countDown();

        Assert.assertTrue(queue.flushPuts(timing.forWaiting().seconds(), TimeUnit.SECONDS));
    }
    finally
    {
        if ( latch.getCount() > 0 )
        {
            latch.countDown();
        }

        CloseableUtils.closeQuietly(queue);
        CloseableUtils.closeQuietly(client);
    }
}
 

@Test
public void testDeserializationInParallel() throws Exception {
  NestedSetStorageEndpoint nestedSetStorageEndpoint =
      Mockito.spy(new InMemoryNestedSetStorageEndpoint());
  NestedSetCache emptyNestedSetCache = mock(NestedSetCache.class);
  NestedSetStore nestedSetStore =
      new NestedSetStore(
          nestedSetStorageEndpoint, emptyNestedSetCache, MoreExecutors.directExecutor());

  ObjectCodecs objectCodecs = createCodecs(nestedSetStore);

  NestedSet<String> subset1 =
      new NestedSetBuilder<String>(Order.STABLE_ORDER).add("a").add("b").build();
  SettableFuture<byte[]> subset1Future = SettableFuture.create();
  NestedSet<String> subset2 =
      new NestedSetBuilder<String>(Order.STABLE_ORDER).add("c").add("d").build();
  SettableFuture<byte[]> subset2Future = SettableFuture.create();
  NestedSet<String> set =
      new NestedSetBuilder<String>(Order.STABLE_ORDER)
          .addTransitive(subset1)
          .addTransitive(subset2)
          .build();

  // We capture the arguments to #put() during serialization, so as to correctly mock results for
  // #get()
  ArgumentCaptor<ByteString> fingerprintCaptor = ArgumentCaptor.forClass(ByteString.class);
  ByteString fingerprint =
      nestedSetStore
          .computeFingerprintAndStore(
              (Object[]) set.getChildren(), objectCodecs.getSerializationContext())
          .fingerprint();
  Mockito.verify(nestedSetStorageEndpoint, Mockito.times(3))
      .put(fingerprintCaptor.capture(), any());
  Mockito.doReturn(subset1Future)
      .when(nestedSetStorageEndpoint)
      .get(fingerprintCaptor.getAllValues().get(0));
  Mockito.doReturn(subset2Future)
      .when(nestedSetStorageEndpoint)
      .get(fingerprintCaptor.getAllValues().get(1));
  when(emptyNestedSetCache.putIfAbsent(any(), any())).thenAnswer(invocation -> null);

  @SuppressWarnings("unchecked")
  ListenableFuture<Object[]> deserializationFuture =
      (ListenableFuture<Object[]>)
          nestedSetStore.getContentsAndDeserialize(
              fingerprint, objectCodecs.getDeserializationContext());
  // At this point, we expect deserializationFuture to be waiting on both of the underlying
  // fetches, which should have both been started.
  assertThat(deserializationFuture.isDone()).isFalse();
  Mockito.verify(nestedSetStorageEndpoint, Mockito.times(3)).get(any());

  // Once the underlying fetches complete, we expect deserialization to complete.
  subset1Future.set(ByteString.copyFrom("mock bytes", Charset.defaultCharset()).toByteArray());
  subset2Future.set(ByteString.copyFrom("mock bytes", Charset.defaultCharset()).toByteArray());
  assertThat(deserializationFuture.isDone()).isTrue();
}
 

protected AbstractMessageListener(RequestResponseMessageBus<T> bus) {
   this(bus, MoreExecutors.directExecutor());
}
 

private RetriableStream<String> newThrottledRetriableStream(Throttle throttle) {
  return new RecordedRetriableStream(
      method, new Metadata(), channelBufferUsed, PER_RPC_BUFFER_LIMIT, CHANNEL_BUFFER_LIMIT,
      MoreExecutors.directExecutor(), fakeClock.getScheduledExecutorService(), RETRY_POLICY,
      HedgingPolicy.DEFAULT, throttle);
}
 

private Execution createExecution(Searcher searcher, Map<Searcher, Searcher> chained) {
    Execution.Context context = new Execution.Context(null, null, null, new RendererRegistry(MoreExecutors.directExecutor()), new SimpleLinguistics());
    return new Execution(chainedAsSearchChain(searcher, chained), context);
}
 

private Execution createExecution(Searcher searcher) {
    Execution.Context context = new Execution.Context(null, null, null, new RendererRegistry(MoreExecutors.directExecutor()), new SimpleLinguistics());
    return new Execution(chainedAsSearchChain(searcher), context);
}
 

private Executor getExecutor() {
  return MoreExecutors.directExecutor();
}
 

private Execution createExecution(Searcher searcher, Map<Searcher, Searcher> chained) {
    Execution.Context context = new Execution.Context(null, null, null, new RendererRegistry(MoreExecutors.directExecutor()), new SimpleLinguistics());
    return new Execution(chainedAsSearchChain(searcher, chained), context);
}
 

public TestSensorDiscoverer(String serviceName) {
  this(serviceName, MoreExecutors.directExecutor());
}
 

private Executor getUiThreadExecutor() {
  return MoreExecutors.directExecutor();
}
 

/**
 * Returns the executor in which to execute a listener.
 * Before handover this *must* be the calling thread, because listeners are free to modify the dataList.
 * After handover it can be any thread in the renderer pool.
 * Note that as some listeners may be set up before handover and executed after, it is possible that some rendering
 * inadvertently work ends up in async data producing threads in some cases.
 */
Executor getExecutor() {
    return beforeHandoverMode ? MoreExecutors.directExecutor() : renderingExecutor;
}