Java 类io.netty.util.concurrent.ImmediateEventExecutor 实例源码

private SslHandler getSslHandler() throws Exception {
  // get SslHandler if it was added to the pipeline
  ArgumentCaptor<ChannelHandler> captor = ArgumentCaptor.forClass(ChannelHandler.class);
  verify(pipeline).addFirst(captor.capture());
  SslHandler sslHandler = (SslHandler) captor.getValue();

  // mock and store the context so we can get the handshake future
  ChannelHandlerContext context = mock(ChannelHandlerContext.class);
  when(context.executor()).thenReturn(ImmediateEventExecutor.INSTANCE);
  when(context.channel()).thenReturn(mock(Channel.class, Answers.RETURNS_MOCKS.get()));

  // add the handler but prevent the handshake from running automatically
  when(channel.isActive()).thenReturn(false);
  sslHandler.handlerAdded(context);

  return sslHandler;
}

@Override
public void abort() {
    abortPending = true;

    final SubscriptionImpl newSubscription = new SubscriptionImpl(
            this, AbortingSubscriber.get(), ImmediateEventExecutor.INSTANCE, false);
    subscriptionUpdater.compareAndSet(this, null, newSubscription);

    final StreamMessage<T> delegate = this.delegate;
    if (delegate != null) {
        delegate.abort();
    } else {
        if (subscription.needsDirectInvocation()) {
            ABORTED_CLOSE.notifySubscriber(subscription, completionFuture());
        } else {
            subscription.executor().execute(
                    () -> ABORTED_CLOSE.notifySubscriber(subscription, completionFuture()));
        }
    }
}

@Override
public void abort() {
    final AbortableSubscriber subscriber = this.subscriber;
    if (subscriber != null) {
        subscriber.abort();
        return;
    }

    final AbortableSubscriber abortable = new AbortableSubscriber(this, AbortingSubscriber.get(),
                                                                  ImmediateEventExecutor.INSTANCE);
    if (!subscriberUpdater.compareAndSet(this, null, abortable)) {
        this.subscriber.abort();
        return;
    }

    abortable.abort();
    abortable.onSubscribe(NoopSubscription.INSTANCE);
}

@Override
public void abort() {
    final SubscriptionImpl currentSubscription = subscription;
    if (currentSubscription != null) {
        cancelOrAbort(false);
        return;
    }

    final SubscriptionImpl newSubscription = new SubscriptionImpl(
            this, AbortingSubscriber.get(), ImmediateEventExecutor.INSTANCE, false);
    if (subscriptionUpdater.compareAndSet(this, null, newSubscription)) {
        // We don't need to invoke onSubscribe() for AbortingSubscriber because it's just a placeholder.
        invokedOnSubscribe = true;
    }
    cancelOrAbort(false);
}

@Test
public void testLeak() {
    final ByteBuf buf = Unpooled.buffer();
    buf.writeCharSequence("foo", StandardCharsets.UTF_8);

    final HttpResponse orig = HttpResponse.of(
            AggregatedHttpMessage.of(HttpStatus.OK,
                                     MediaType.PLAIN_TEXT_UTF_8,
                                     new ByteBufHttpData(buf, true)));
    final HttpEncodedResponse encoded = new HttpEncodedResponse(
            orig, HttpEncodingType.DEFLATE, mediaType -> true, 1);

    // Drain the stream.
    encoded.subscribe(NoopSubscriber.get(), ImmediateEventExecutor.INSTANCE);

    // 'buf' should be released.
    assertThat(buf.refCnt()).isZero();
}

@Test
public void testLateAbortWithSubscriber() throws Exception {
    final DeferredStreamMessage<Object> m = new DeferredStreamMessage<>();
    final DefaultStreamMessage<Object> d = new DefaultStreamMessage<>();
    @SuppressWarnings("unchecked")
    final Subscriber<Object> subscriber = mock(Subscriber.class);

    m.subscribe(subscriber, ImmediateEventExecutor.INSTANCE);
    m.delegate(d);
    verify(subscriber).onSubscribe(any());

    m.abort();
    verify(subscriber, times(1)).onError(isA(AbortedStreamException.class));

    assertAborted(m);
    assertAborted(d);
}

@Test
public void shouldCancelCommandOnQueueSingleFailure() throws Exception {

    Command<String, String, String> commandMock = mock(Command.class);

    RuntimeException exception = new RuntimeException();
    when(commandMock.getOutput()).thenThrow(exception);

    ChannelPromise channelPromise = new DefaultChannelPromise(null, ImmediateEventExecutor.INSTANCE);
    try {
        sut.write(context, commandMock, channelPromise);
        fail("Missing RuntimeException");
    } catch (RuntimeException e) {
        assertThat(e).isSameAs(exception);
    }

    assertThat(disconnectedBuffer).isEmpty();
    verify(commandMock).completeExceptionally(exception);
}

@Test
public void shouldCancelCommandOnQueueBatchFailure() throws Exception {

    Command<String, String, String> commandMock = mock(Command.class);

    RuntimeException exception = new RuntimeException();
    when(commandMock.getOutput()).thenThrow(exception);

    ChannelPromise channelPromise = new DefaultChannelPromise(null, ImmediateEventExecutor.INSTANCE);
    try {
        sut.write(context, Arrays.asList(commandMock), channelPromise);
        fail("Missing RuntimeException");
    } catch (RuntimeException e) {
        assertThat(e).isSameAs(exception);
    }

    assertThat(disconnectedBuffer).isEmpty();
    verify(commandMock).completeExceptionally(exception);
}

@Test
public void testList_01() throws InterruptedException {
    MetadataRepository repository = mockRepository();
    MetadataService mds = new DefaultMetadataServiceImpl(repository);

    ObjectRequest request = new ObjectRequest(httpRequest(HttpMethod.GET, "/owner/index/parent/path"));

    mds.list(ImmediateEventExecutor.INSTANCE, request).addListener(
            new GenericFutureListener<Future<? super ObjectResponse>>() {
                @Override
                public void operationComplete(Future<? super ObjectResponse> future) throws Exception {
                    if (future.isSuccess()) {
                        log.info("fetched: {}", future.get());
                    } else {
                        log.debug("failed: {}", future.cause());
                    }
                }
            }
    ).sync();

}

/**
 * Sends a message to this pipe. Returns a {@link Future} that is completed
 * when the message is received.
 * <p>
 * If the pipe is closed then this will return a failed future.</p>
 *
 * @param message the message to send to the pipe
 * @return a {@link Future} that is satisfied when the message is received,
 * or a failed future if the pipe is closed.
 * @throws NullPointerException  if the message is {@code null}.
 * @throws IllegalStateException if the message could not be added to the queue for some reason.
 * @see #receive()
 */
public Future<Void> send(T message) {
    Objects.requireNonNull(message, "msg");

    Promise<T> receivePromise;

    synchronized (this) {
        if (closed) {
            return CLOSED_FUTURE;
        }

        receivePromise = receiveQueue.poll();
        if (receivePromise == null) {
            Promise<Void> sendPromise = ImmediateEventExecutor.INSTANCE.newPromise();
            sendQueue.add(new Node(message, sendPromise));
            return sendPromise;
        }
    }

    receivePromise.setSuccess(message);
    return SENT_FUTURE;
}

/**
 * Receives a message from this pipe.
 * <p>
 * If the pipe is closed then this will return a failed future.</p>
 */
public Future<T> receive() {
    Node node;

    synchronized (this) {
        node = sendQueue.poll();
        if (node == null) {
            if (closed) {
                return ImmediateEventExecutor.INSTANCE.newFailedFuture(PIPE_CLOSED);
            }

            Promise<T> promise = ImmediateEventExecutor.INSTANCE.newPromise();
            receiveQueue.add(promise);
            return promise;
        }
    }

    node.promise.setSuccess(null);
    return ImmediateEventExecutor.INSTANCE.newSucceededFuture(node.message);
}

@Test
public void itClosesTheUnderlyingChannel() {
  DefaultChannelPromise channelPromise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
  when(channel.close()).thenReturn(channelPromise);

  CompletableFuture<Void> f = session.close();
  channelPromise.setSuccess();

  assertThat(f.isDone());
}

private void assertExceptionsFiredOnFailure() throws Exception {
  // get the listener added when the channel was written to
  ArgumentCaptor<ChannelFutureListener> captor = ArgumentCaptor.forClass(ChannelFutureListener.class);
  verify(writeFuture, atLeast(1)).addListener(captor.capture());
  ChannelFutureListener addedListener = captor.getValue();

  // tell the listener the write failed
  DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
  promise.setFailure(new Exception());
  addedListener.operationComplete(promise);

  verify(pipeline).fireExceptionCaught(promise.cause());
}

public WebSocketServerThread(Settings settings) {
    this.PORT = settings.httpPort;
    this.SSL = false; // TODO: support ssl?

    this.blockBridge = null;
    this.playersBridge = null;
    this.webPlayerBridge = null;

    this.allUsersGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);

    this.settings = settings;
}

@Override
protected void startUp() throws Exception {
  channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
  EventLoopGroup bossGroup = new NioEventLoopGroup(NUM_BOSS_THREADS,
                                                   new ThreadFactoryBuilder()
                                                     .setDaemon(true).setNameFormat("boss-thread").build());
  EventLoopGroup workerGroup = new NioEventLoopGroup(NUM_WORKER_THREADS,
                                                     new ThreadFactoryBuilder()
                                                       .setDaemon(true).setNameFormat("worker-thread#%d").build());

  bootstrap = new ServerBootstrap()
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        channelGroup.add(ch);
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("codec", new HttpServerCodec());
        pipeline.addLast("compressor", new HttpContentCompressor());
        pipeline.addLast("aggregator", new HttpObjectAggregator(MAX_INPUT_SIZE));
        pipeline.addLast("handler", new ReportHandler());
      }
    });

  Channel serverChannel = bootstrap.bind(new InetSocketAddress(host, 0)).sync().channel();
  channelGroup.add(serverChannel);

  bindAddress = (InetSocketAddress) serverChannel.localAddress();
  url = URI.create(String.format("http://%s:%d", host, bindAddress.getPort())).toURL();

  LOG.info("Tracker service started at {}", url);
}

@Test
@SuppressWarnings("unchecked")
public void testDestroy() throws Exception {
    Future<String> future = new SucceededFuture<String>(ImmediateEventExecutor.INSTANCE, "test");
    when(eventLoopGroup.shutdownGracefully()).thenReturn((Future) future);
    jannelClient.destroy();

    verify(eventLoopGroup).shutdownGracefully();
}

public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
    this.camelContext = camelContext;
    this.configuration = configuration;
    this.pipelineFactory = pipelineFactory;

    this.allChannels = configuration.getChannelGroup() != null
        ? configuration.getChannelGroup()
        : new DefaultChannelGroup(SingleTCPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}

public void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
    this.threadFactory = threadFactory;
    this.configuration = configuration;
    this.pipelineFactory = pipelineFactory;

    this.allChannels = configuration.getChannelGroup() != null
        ? configuration.getChannelGroup()
        : new DefaultChannelGroup(SingleTCPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}

@Override
protected void checkDeadLock() {
    EventExecutor e = executor();
    if (e != null && e != ImmediateEventExecutor.INSTANCE && e.inEventLoop()) {
        throw new BlockingOperationException();
    }
}

private static void invoke0(THttpService service, HttpData content,
                            CompletableFuture<HttpData> promise) throws Exception {

    final ServiceRequestContext ctx = mock(ServiceRequestContext.class);
    when(ctx.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
    final DefaultRequestLog reqLogBuilder = new DefaultRequestLog(ctx);

    when(ctx.blockingTaskExecutor()).thenReturn(ImmediateEventExecutor.INSTANCE);
    when(ctx.log()).thenReturn(reqLogBuilder);
    when(ctx.logBuilder()).thenReturn(reqLogBuilder);
    doNothing().when(ctx).invokeOnEnterCallbacks();
    doNothing().when(ctx).invokeOnExitCallbacks();

    final HttpRequestWriter req = HttpRequest.streaming(HttpHeaders.of(HttpMethod.POST, "/"));

    req.write(content);
    req.close();

    final HttpResponse res = service.serve(ctx, req);
    res.aggregate().handle(voidFunction((aReq, cause) -> {
        if (cause == null) {
            if (aReq.headers().status().code() == 200) {
                promise.complete(aReq.content());
            } else {
                promise.completeExceptionally(new AssertionError(
                        aReq.headers().status() + ", " +
                        aReq.content().toString(StandardCharsets.UTF_8)));
            }
        } else {
            promise.completeExceptionally(cause);
        }
    })).exceptionally(CompletionActions::log);
}

@Override
public final void abort() {
    final SubscriptionImpl currentSubscription = subscription;
    if (currentSubscription != null) {
        cancelOrAbort(false);
        return;
    }

    final SubscriptionImpl newSubscription = new SubscriptionImpl(
            this, AbortingSubscriber.get(), ImmediateEventExecutor.INSTANCE, false);
    subscriptionUpdater.compareAndSet(this, null, newSubscription);
    cancelOrAbort(false);
}

@Test
public void testEarlyAbortWithSubscriber() throws Exception {
    final DeferredStreamMessage<Object> m = new DeferredStreamMessage<>();
    m.subscribe(mock(Subscriber.class), ImmediateEventExecutor.INSTANCE);
    m.abort();
    assertAborted(m);

    final DefaultStreamMessage<Object> d = new DefaultStreamMessage<>();
    m.delegate(d);
    assertAborted(d);
}

@Test
public void testEarlySubscription() throws Exception {
    final DeferredStreamMessage<Object> m = new DeferredStreamMessage<>();
    final DefaultStreamMessage<Object> d = new DefaultStreamMessage<>();
    @SuppressWarnings("unchecked")
    final Subscriber<Object> subscriber = mock(Subscriber.class);

    m.subscribe(subscriber, ImmediateEventExecutor.INSTANCE);
    assertFailedSubscription(m, IllegalStateException.class);

    m.delegate(d);
    verify(subscriber).onSubscribe(any());
}

@Test
public void testLateSubscription() throws Exception {
    final DeferredStreamMessage<Object> m = new DeferredStreamMessage<>();
    final DefaultStreamMessage<Object> d = new DefaultStreamMessage<>();

    m.delegate(d);

    @SuppressWarnings("unchecked")
    final Subscriber<Object> subscriber = mock(Subscriber.class);

    m.subscribe(subscriber, ImmediateEventExecutor.INSTANCE);
    verify(subscriber).onSubscribe(any());

    assertFailedSubscription(m, IllegalStateException.class);
}

@Test
public void subscribeTwice() {
    @SuppressWarnings("unchecked")
    final StreamMessage<String> publisher = mock(StreamMessage.class);
    when(publisher.completionFuture()).thenReturn(new CompletableFuture<>());

    final StreamMessageDuplicator duplicator = new StreamMessageDuplicator(publisher);

    @SuppressWarnings("unchecked")
    final ArgumentCaptor<StreamMessageProcessor<String>> processorCaptor =
            ArgumentCaptor.forClass(StreamMessageProcessor.class);

    verify(publisher).subscribe(processorCaptor.capture(), eq(ImmediateEventExecutor.INSTANCE), eq(true));

    verify(publisher).subscribe(any(), eq(ImmediateEventExecutor.INSTANCE), eq(true));
    final Subscriber<String> subscriber1 = subscribeWithMock(duplicator.duplicateStream());
    final Subscriber<String> subscriber2 = subscribeWithMock(duplicator.duplicateStream());
    // Publisher's subscribe() is not invoked when a new subscriber subscribes.
    verify(publisher).subscribe(any(), eq(ImmediateEventExecutor.INSTANCE), eq(true));

    final StreamMessageProcessor<String> processor = processorCaptor.getValue();

    // Verify that the propagated triggers onSubscribe().
    verify(subscriber1, never()).onSubscribe(any());
    verify(subscriber2, never()).onSubscribe(any());
    processor.onSubscribe(mock(Subscription.class));
    verify(subscriber1).onSubscribe(any(DownstreamSubscription.class));
    verify(subscriber2).onSubscribe(any(DownstreamSubscription.class));
    duplicator.close();
}

@Test
public void lastDuplicateStream() {
    final DefaultStreamMessage<ByteBuf> publisher = new DefaultStreamMessage<>();
    final ByteBufDuplicator duplicator = new ByteBufDuplicator(publisher);

    duplicator.duplicateStream().subscribe(new ByteBufSubscriber(), ImmediateEventExecutor.INSTANCE);
    duplicator.duplicateStream(true).subscribe(new ByteBufSubscriber(), ImmediateEventExecutor.INSTANCE);

    // duplicateStream() is not allowed anymore.
    assertThatThrownBy(duplicator::duplicateStream).isInstanceOf(IllegalStateException.class);

    final ByteBuf[] bufs = new ByteBuf[30];
    for (int i = 0; i < 30; i++) {
        final ByteBuf buf = newUnpooledBuffer();
        bufs[i] = buf;
        assertThat(publisher.write(buf)).isTrue();  // Removing internal caches happens when i = 25
        assertThat(buf.refCnt()).isOne();
    }

    for (int i = 0; i < 25; i++) {  // first 25 signals are removed from the queue.
        assertThat(bufs[i].refCnt()).isZero();
    }
    for (int i = 25; i < 30; i++) {  // rest of them are still in the queue.
        assertThat(bufs[i].refCnt()).isOne();
        bufs[i].release();
    }
}

@Test
@SuppressWarnings("deprecation")
public void makeContextAwareFutureListener() {
    RequestContext context = createContext();
    Promise<String> promise = new DefaultPromise<>(ImmediateEventExecutor.INSTANCE);
    promise.addListener(context.makeContextAware((FutureListener<String>) f -> {
        assertCurrentContext(context);
        assertDepth(1);
        assertThat(f.getNow()).isEqualTo("success");
    }));
    promise.setSuccess("success");
}

@Test
@SuppressWarnings("deprecation")
public void makeContextAwareChannelFutureListener() {
    RequestContext context = createContext();
    ChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
    promise.addListener(context.makeContextAware((ChannelFutureListener) f -> {
        assertCurrentContext(context);
        assertDepth(1);
        assertThat(f.getNow()).isNull();
    }));
    promise.setSuccess(null);
}

@Override
public Promise<Boolean> release(EventExecutorGroup eventLoopGroup, long quietPeriod, long timeout, TimeUnit unit) {
    DefaultPromise<Boolean> result = new DefaultPromise<Boolean>(ImmediateEventExecutor.INSTANCE);
    result.setSuccess(true);

    return result;
}

@Override
protected void checkDeadLock() {
    EventExecutor e = executor();
    if (e != null && e != ImmediateEventExecutor.INSTANCE && e.inEventLoop()) {
        throw new BlockingOperationException();
    }
}

@Test(expected = IllegalStateException.class)
public void expectAfterArmed() {
    Futures.PromiseAggregator<Boolean, Promise<Boolean>> sut = new Futures.PromiseAggregator<>(new DefaultPromise<>(
            ImmediateEventExecutor.INSTANCE));
    sut.arm();

    sut.expectMore(1);
}

@Test(expected = IllegalStateException.class)
public void armTwice() {
    Futures.PromiseAggregator<Boolean, Promise<Boolean>> sut = new Futures.PromiseAggregator<>(new DefaultPromise<>(
            ImmediateEventExecutor.INSTANCE));
    sut.arm();
    sut.arm();
}

@Override
public Promise<Boolean> release(EventExecutorGroup eventLoopGroup, long quietPeriod, long timeout, TimeUnit unit) {
    DefaultPromise<Boolean> result = new DefaultPromise<>(ImmediateEventExecutor.INSTANCE);
    result.setSuccess(true);

    return result;
}

@Before
public void before() {

    DefaultClientResources clientResources = DefaultClientResources.builder()
            .eventExecutorGroup(ImmediateEventExecutor.INSTANCE)
            .eventLoopGroupProvider(new DefaultEventLoopGroupProvider(1))
            .commandLatencyCollectorOptions(DefaultCommandLatencyCollectorOptions.disabled()).build();

    client = RedisClusterClient.create(clientResources, RedisURI.create("localhost", 7379));
}

@Override
protected void checkDeadLock() {
    EventExecutor e = executor();
    if (e != null && e != ImmediateEventExecutor.INSTANCE && e.inEventLoop()) {
        throw new BlockingOperationException();
    }
}

@Override
public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
    shutdown();
    if (isShutdown()) {
        return ImmediateEventExecutor.INSTANCE.newSucceededFuture(null);
    } else {
        return ImmediateEventExecutor.INSTANCE.newFailedFuture(new TimeoutException());
    }
}

@Test
public void testEncodeMethodCallFailure() throws InvalidProtocolBufferException,
    InterruptedException {
  Channel channel = Mockito.mock(Channel.class);
  ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);

  DefaultChannelPromise failure = new DefaultChannelPromise(
      channel, ImmediateEventExecutor.INSTANCE);
  failure.setFailure(new Exception("OMGWTF"));
  Mockito.when(channel.writeAndFlush(captor.capture())).thenReturn(failure);

  RpcClientHandler handler = new RpcClientHandler();
  RpcClient client = new RpcClient(channel, handler, new NullClientLogger());

  ClientMethod<TimeResponse> method = Mockito.mock(ClientMethod.class);
  Mockito.when(method.serviceName()).thenReturn("TimeService");
  Mockito.when(method.name()).thenReturn("GetTime");
  Mockito.when(method.outputParser()).thenReturn(TimeResponse.PARSER);

  final CountDownLatch latch = new CountDownLatch(1);
  FutureCallback<TimeResponse> callback = new FutureCallback<TimeResponse>() {
    @Override
    public void onSuccess(@Nullable TimeResponse result) {
    }

    @Override
    public void onFailure(Throwable t) {
      Assert.assertEquals("OMGWTF", t.getMessage());
      latch.countDown();
    }
  };

  ListenableFuture<TimeResponse> future = client.encodeMethodCall(
      method, TimeRequest.newBuilder().setTimezone("UTC").build());

  Futures.addCallback(future, callback);
  latch.await(5, TimeUnit.SECONDS);

  Assert.assertEquals(0, handler.inFlightRequests().size());
}

public static AutoCloseableEventExecutor immediateEventExecutor() {
    return createCloseableProxy(new CloseableEventExecutorMixin(ImmediateEventExecutor.INSTANCE));
}

public SingleUDPNettyServerBootstrapFactory() {
    this.allChannels = new DefaultChannelGroup(SingleUDPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}

public NettyTelnetBootstrap() {
  this.group = new NioEventLoopGroup();
  this.channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
}