Java 类io.netty.channel.EventLoop 实例源码

private <T> void scheduleTimeout(CompletableFuture<T> result, long timeoutMillis) {
    pendingFutures.add(result);
    if (isServerStopping()) {
        pendingFutures.remove(result);
        return;
    }

    final ScheduledFuture<?> timeoutFuture;
    if (timeoutMillis > 0) {
        final EventLoop eventLoop = RequestContext.current().eventLoop();
        timeoutFuture = eventLoop.schedule(() -> result.completeExceptionally(CANCELLATION_EXCEPTION),
                                           timeoutMillis, TimeUnit.MILLISECONDS);
    } else {
        timeoutFuture = null;
    }

    result.whenComplete((revision, cause) -> {
        if (timeoutFuture != null) {
            timeoutFuture.cancel(true);
        }
        pendingFutures.remove(result);
    });
}

private EventLoop mapToThread(int affinity, HandlerRegistration handler) {
    EventLoopGroup group;

    // Check if a dedicated thread pool is defined for this protocol.
    if (handler.config().getEventLoop() == null) {
        // Use core thread pool.
        group = coreEventLoopGroup;
    } else {
        // Use dedicated thread pool.
        group = handler.config().getEventLoop();
    }

    List<EventLoop> eventLoops = new ArrayList<>();

    // Assumes that the same group always returns its event loops in the same order.
    for (Iterator<EventExecutor> it = group.iterator(); it.hasNext(); ) {
        eventLoops.add((EventLoop)it.next());
    }

    return eventLoops.get(Utils.mod(affinity, eventLoops.size()));
}

/**
 * Executes the task using the provided event loop or falls back to {@link AsyncUtils#fallbackExecutor()} if event loop is {@link
 * EventLoop#isShuttingDown() shut down}.
 *
 * @param eventLoop Event loop.
 * @param task Task.
 */
public static void runAtAllCost(EventLoop eventLoop, Runnable task) {
    assert eventLoop != null : "Event loop is null.";
    assert task != null : "Task is null.";

    boolean notified = false;

    // Try to execute via event loop.
    if (!eventLoop.isShuttingDown()) {
        try {
            eventLoop.execute(task);

            notified = true;
        } catch (RejectedExecutionException e) {
            // No-op.
        }
    }

    // If couldn't notify via event loop then use the fallback executor.
    if (!notified) {
        AsyncUtils.fallbackExecutor().execute(task);
    }
}

@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
    if (!channelFuture.isSuccess()) {
        channelFuture.channel().close();

        if (count.incrementAndGet() < MAX_RETRY) {
            final EventLoop loop = channelFuture.channel().eventLoop();

            loop.schedule(() -> {
                controller.connectRetry(this.ip, this.port, this);
            }, 1L, TimeUnit.SECONDS);
        } else {
            log.info("Connection to the ovsdb {}:{} failed",
                     this.ip.toString(), this.port.toString());
        }
    } else {
        handleNewNodeConnection(channelFuture.channel());
    }
}

/**
 * Constructor
 *
 * @param eventLoop           for call
 * @param connectId           connection id
 * @param md                  the method descriptor
 * @param param               parameters to send to Server
 * @param controller          controller for response
 * @param responseDefaultType the default response type
 */
public AsyncCall(EventLoop eventLoop, int connectId, Descriptors.MethodDescriptor md, Message
    param, PayloadCarryingRpcController controller, Message responseDefaultType,
    MetricsConnection.CallStats callStats) {
  super(eventLoop);

  this.id = connectId;

  this.method = md;
  this.param = param;
  this.controller = controller;
  this.responseDefaultType = responseDefaultType;

  this.startTime = EnvironmentEdgeManager.currentTime();
  this.rpcTimeout = controller.hasCallTimeout() ? controller.getCallTimeout() : 0;
  this.callStats = callStats;
}

protected Optional<CircuitBreaker<HttpResponse>> getCircuitBreaker(
    DownstreamRequestFirstChunkInfo downstreamReqFirstChunkInfo, ChannelHandlerContext ctx
) {
    if (downstreamReqFirstChunkInfo == null || downstreamReqFirstChunkInfo.disableCircuitBreaker)
        return Optional.empty();

    // Circuit breaking is enabled for this call. So we return the custom one specified or use the default one if a
    //      custom one is not specified.
    if (downstreamReqFirstChunkInfo.customCircuitBreaker.isPresent())
        return downstreamReqFirstChunkInfo.customCircuitBreaker;

    // No custom circuit breaker. Use the default for the given request's host.
    EventLoop nettyEventLoop = ctx.channel().eventLoop();
    CircuitBreaker<Integer> defaultStatusCodeCircuitBreaker = getDefaultHttpStatusCodeCircuitBreakerForKey(
        downstreamReqFirstChunkInfo.host, Optional.ofNullable(nettyEventLoop), Optional.ofNullable(nettyEventLoop)
    );
    return Optional.of(
        new CircuitBreakerDelegate<>(
            defaultStatusCodeCircuitBreaker,
            httpResponse -> (httpResponse == null ? null : httpResponse.getStatus().code())
        )
    );
}

protected Optional<CircuitBreaker<Response>> getCircuitBreaker(RequestBuilderWrapper requestBuilderWrapper) {
    if (requestBuilderWrapper.disableCircuitBreaker)
        return Optional.empty();

    // Circuit breaking is enabled for this call. So we return the custom one specified or use the default one if a
    //      custom one is not specified.
    if (requestBuilderWrapper.customCircuitBreaker.isPresent())
        return requestBuilderWrapper.customCircuitBreaker;

    // No custom circuit breaker. Use the default for the given request's host.
    Uri uri = Uri.create(requestBuilderWrapper.url);
    String host = uri.getHost();
    EventLoop nettyEventLoop = requestBuilderWrapper.getCtx() == null
                               ? null
                               : requestBuilderWrapper.getCtx().channel().eventLoop();
    CircuitBreaker<Integer> defaultStatusCodeCircuitBreaker = getDefaultHttpStatusCodeCircuitBreakerForKey(
        host, Optional.ofNullable(nettyEventLoop), Optional.ofNullable(nettyEventLoop)
    );
    return Optional.of(
        new CircuitBreakerDelegate<>(
            defaultStatusCodeCircuitBreaker, response -> (response == null ? null : response.getStatusCode())
        )
    );
}

@Before
public void beforeMethod() {
    helperSpy = spy(new AsyncHttpClientHelper());
    channelMock = mock(Channel.class);
    ctxMock = mock(ChannelHandlerContext.class);
    stateAttributeMock = mock(Attribute.class);
    state = new HttpProcessingState();
    eventLoopMock = mock(EventLoop.class);
    signatureCalculator = mock(SignatureCalculator.class);
    doReturn(channelMock).when(ctxMock).channel();
    doReturn(stateAttributeMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(state).when(stateAttributeMock).get();
    doReturn(eventLoopMock).when(channelMock).eventLoop();

    handlerWithTracingAndMdcDummyExample = new AsyncCompletionHandlerWithTracingAndMdcSupport<>(
        null, null, false, null, null, null, null, null
    );

    resetTracingAndMdc();
}

/**
 * Dispatches an AddressedMessage to its target handler using an EventExecutor.
 *
 * @param msg AddressedMessage to dispatch.
 * @return {@code true} if the Message was forwarded to at least one MessageHandler.
 */
public boolean dispatch(final Message.AddressedMessage msg) {
    Set<MessageHandler> handlers = mappings.get(RoutingKey.forMessage(msg));
    final EventLoop executor = getEventLoop();
    Log.v(TAG, "DISPATCH " + msg + " to " + handlers + " using " + executor);
    for (final MessageHandler handler : handlers) {
        executor.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    handler.handle(msg);
                } catch (Exception e) {
                    Log.e(TAG, "Handler " + handler + " crashed while handling message " + msg
                            + " with Exception " + Log.getStackTraceString(e));
                }
            }
        });
    }
    return !handlers.isEmpty();
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    LOG.info(">>> channelUnregistered");
    if (unrecognizedName) {
        LOG.info(">>> unrecognizedName retry");
        final EventLoop loop = ctx.channel().eventLoop();
        loop.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    client.retry(loop);
                } catch (InterruptedException e) {
                    LOG.info(">>> retry interrupted, shutdown");
                    client.stop();
                }
            }
        });
    } else {
        LOG.info(">>> shutdown sucessfully");
        client.stop();
    }
}

public Future<Boolean> addAsync(final V value) {
    EventLoop loop = commandExecutor.getConnectionManager().getGroup().next();
    final Promise<Boolean> promise = loop.newPromise();

    loop.execute(new Runnable() {
        @Override
        public void run() {
            try {
                boolean result = add(value);
                promise.setSuccess(result);
            } catch (Exception e) {
                promise.setFailure(e);
            }
        }
    });

    return promise;
}

@Override
public ChannelFuture shutdownOutput(final ChannelPromise future) {
    EventLoop loop = eventLoop();
    if (loop.inEventLoop()) {
        try {
            socket.shutdownOutput();
            future.setSuccess();
        } catch (Throwable t) {
            future.setFailure(t);
        }
    } else {
        loop.execute(new Runnable() {
            @Override
            public void run() {
                shutdownOutput(future);
            }
        });
    }
    return future;
}

@Override
public ChannelFuture shutdownOutput(final ChannelPromise promise) {
    Executor closeExecutor = ((NioSocketChannelUnsafe) unsafe()).closeExecutor();
    if (closeExecutor != null) {
        closeExecutor.execute(new OneTimeTask() {
            @Override
            public void run() {
                shutdownOutput0(promise);
            }
        });
    } else {
        EventLoop loop = eventLoop();
        if (loop.inEventLoop()) {
            shutdownOutput0(promise);
        } else {
            loop.execute(new OneTimeTask() {
                @Override
                public void run() {
                    shutdownOutput0(promise);
                }
            });
        }
    }
    return promise;
}

final void clearEpollIn() {
    // Only clear if registered with an EventLoop as otherwise
    if (isRegistered()) {
        final EventLoop loop = eventLoop();
        final AbstractEpollUnsafe unsafe = (AbstractEpollUnsafe) unsafe();
        if (loop.inEventLoop()) {
            unsafe.clearEpollIn0();
        } else {
            // schedule a task to clear the EPOLLIN as it is not safe to modify it directly
            loop.execute(new OneTimeTask() {
                @Override
                public void run() {
                    if (!config().isAutoRead() && !unsafe.readPending) {
                        // Still no read triggered so clear it now
                        unsafe.clearEpollIn0();
                    }
                }
            });
        }
    } else  {
        // The EventLoop is not registered atm so just update the flags so the correct value
        // will be used once the channel is registered
        flags &= ~readFlag;
    }
}

@Override
public ChannelFuture shutdownOutput(final ChannelPromise promise) {
    Executor closeExecutor = ((EpollSocketChannelUnsafe) unsafe()).closeExecutor();
    if (closeExecutor != null) {
        closeExecutor.execute(new OneTimeTask() {
            @Override
            public void run() {
                shutdownOutput0(promise);
            }
        });
    } else {
        EventLoop loop = eventLoop();
        if (loop.inEventLoop()) {
            shutdownOutput0(promise);
        } else {
            loop.execute(new OneTimeTask() {
                @Override
                public void run() {
                    shutdownOutput0(promise);
                }
            });
        }
    }
    return promise;
}

private ClientRequestContext newContext(HttpMethod method, HttpRequest req) {
    final ReleasableHolder<EventLoop> eventLoop = factory().acquireEventLoop(endpoint);
    final ClientRequestContext ctx = new DefaultClientRequestContext(
            eventLoop.get(),
            meterRegistry,
            sessionProtocol,
            endpoint,
            method,
            uri().getRawPath(),
            uri().getRawQuery(),
            null,
            options(),
            req);
    ctx.log().addListener(log -> eventLoop.release(), RequestLogAvailability.COMPLETE);
    return ctx;
}

@Override
public Future<Void> release(final K key, final Channel channel, final Promise<Void> promise) {
    requireNonNull(key, "key");
    requireNonNull(channel, "channel");
    requireNonNull(promise, "promise");

    try {
        EventLoop loop = channel.eventLoop();
        if (loop.inEventLoop()) {
            doRelease(key, channel, promise);
        } else {
            loop.execute(() -> doRelease(key, channel, promise));
        }
    } catch (Throwable cause) {
        closeAndFail(channel, cause, promise);
    }
    return promise;
}

private HttpResponse execute(@Nullable EventLoop eventLoop, HttpRequest req) {
    final String concatPaths = concatPaths(uri().getRawPath(), req.path());
    req.path(concatPaths);

    final PathAndQuery pathAndQuery = PathAndQuery.parse(concatPaths);
    if (pathAndQuery == null) {
        req.abort();
        return HttpResponse.ofFailure(new IllegalArgumentException("invalid path: " + concatPaths));
    }

    return execute(eventLoop, req.method(), pathAndQuery.path(), pathAndQuery.query(), null, req, cause -> {
        final HttpResponseWriter res = HttpResponse.streaming();
        res.close(cause);
        return res;
    });
}

/**
 * Executes the specified {@link Request} via {@link #delegate()}.
 *
 * @param eventLoop the {@link EventLoop} to execute the {@link Request}
 * @param method the method of the {@link Request}
 * @param path the path part of the {@link Request} URI
 * @param query the query part of the {@link Request} URI
 * @param fragment the fragment part of the {@link Request} URI
 * @param req the {@link Request}
 * @param fallback the fallback response {@link Function} to use when
 *                 {@link Client#execute(ClientRequestContext, Request)} of {@link #delegate()} throws
 */
protected final O execute(@Nullable EventLoop eventLoop,
                          HttpMethod method, String path, @Nullable String query, @Nullable String fragment,
                          I req, Function<Throwable, O> fallback) {

    final ClientRequestContext ctx;
    if (eventLoop == null) {
        final ReleasableHolder<EventLoop> releasableEventLoop = factory().acquireEventLoop(endpoint);
        ctx = new DefaultClientRequestContext(
                releasableEventLoop.get(), meterRegistry, sessionProtocol, endpoint,
                method, path, query, fragment, options(), req);
        ctx.log().addListener(log -> releasableEventLoop.release(), RequestLogAvailability.COMPLETE);
    } else {
        ctx = new DefaultClientRequestContext(eventLoop, meterRegistry, sessionProtocol, endpoint,
                                              method, path, query, fragment, options(), req);
    }

    try (SafeCloseable ignored = RequestContext.push(ctx)) {
        return delegate().execute(ctx, req);
    } catch (Throwable cause) {
        ctx.logBuilder().endResponse(cause);
        return fallback.apply(cause);
    }
}

/**
 * A simple case.
 * (acquire, release) * 3.
 */
@Test
public void acquireAndRelease() {
    final EventLoopScheduler s = new EventLoopScheduler(group);
    final Entry e0 = s.acquire(endpoint);
    final EventLoop loop = e0.get();
    assertThat(e0.id()).isZero();
    assertThat(e0.activeRequests()).isEqualTo(1);
    e0.release();
    assertThat(e0.activeRequests()).isZero();

    for (int i = 0; i < 2; i++) {
        final Entry e0again = s.acquire(endpoint);
        assertThat(e0again).isSameAs(e0);
        assertThat(e0again.id()).isZero();
        assertThat(e0again.activeRequests()).isEqualTo(1);
        assertThat(e0again.get()).isSameAs(loop);
        e0again.release();
    }
}

@Test
public void rpc() {
    final MeterRegistry registry = PrometheusMeterRegistries.newRegistry();
    final ClientRequestContext ctx = new DefaultClientRequestContext(
            mock(EventLoop.class), registry, SessionProtocol.H2C,
            Endpoint.of("example.com", 8080), HttpMethod.POST, "/bar", null, null,
            ClientOptions.DEFAULT, HttpRequest.of(HttpMethod.POST, "/bar"));

    final MeterIdPrefixFunction meterIdPrefixFunction = MeterIdPrefixFunction.ofDefault("bar");

    ctx.logBuilder().startRequest(mock(Channel.class), SessionProtocol.H2C, "example.com");
    RequestMetricSupport.setup(ctx, meterIdPrefixFunction);

    ctx.logBuilder().requestHeaders(HttpHeaders.of(HttpMethod.POST, "/bar"));
    ctx.logBuilder().requestContent(new DefaultRpcRequest(Object.class, "baz"), null);

    assertThat(registry.find("bar.activeRequests")
                       .tags("method", "baz")
                       .value(Statistic.Count, 1).meter()).isPresent();
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    LOG.info(">>> channelUnregistered");
    if (unrecognizedName) {
        LOG.info(">>> unrecognizedName retry");
        final EventLoop loop = ctx.channel().eventLoop();
        loop.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    client.retry(loop);
                } catch (InterruptedException e) {
                    LOG.info(">>> retry interrupted, shutdown");
                    client.stop();
                }
            }
        });
    } else {
        LOG.info(">>> shutdown sucessfully");
        client.stop();
    }
}

public NedisClientImpl(Channel channel, NedisClientPool pool) {
    this.channel = channel;
    this.pool = pool;
    EventLoop eventLoop = channel.eventLoop();
    this.listConverter = PromiseConverter.toList(eventLoop);
    this.booleanConverter = PromiseConverter.toBoolean(eventLoop);
    this.bytesConverter = PromiseConverter.toBytes(eventLoop);
    this.doubleConverter = PromiseConverter.toDouble(eventLoop);
    this.longConverter = PromiseConverter.toLong(eventLoop);
    this.objectConverter = PromiseConverter.toObject(eventLoop);
    this.stringConverter = PromiseConverter.toString(eventLoop);
    this.voidConverter = PromiseConverter.toVoid(eventLoop);
    this.arrayScanResultConverter = PromiseConverter.toArrayScanResult(eventLoop);
    this.mapConverter = PromiseConverter.toMap(eventLoop);
    this.hashScanResultConverter = PromiseConverter.toHashScanResult(eventLoop);
    this.setConverter = PromiseConverter.toSet(eventLoop);
    this.sortedSetEntryListConverter = PromiseConverter.toSortedSetEntryList(eventLoop);
    this.sortedSetScanResultConverter = PromiseConverter.toSortedSetScanResult(eventLoop);
    this.booleanListConverter = PromiseConverter.toBooleanList(eventLoop);
    this.objectListConverter = PromiseConverter.toObjectList(eventLoop);
}

/**
 * A Netty Channel is paused and don’t accept any command temporarily.
 *
 * @param channelProxy the specified channel proxy
 */
private void pauseChannelProxy(final ChannelProxy channelProxy) {
    channelProxy.paused();
    log.info("Pause a channel proxy from pool. channel proxy: {}", channelProxy);

    if (false == channelProxy.hasWaitingRequests()) {
        return;
    }

    final Channel channel = channelProxy.getChannel();
    EventLoop eventLoop = channel.eventLoop();
    eventLoop.schedule(new Runnable() {
        @Override
        public void run() {
            // cancel all waiting requests belong to this channel
            channelProxy.cancelWaitingRequests();
        }
    }, Constants.CANCEL_WAITING_REQUEST_DELAY, TimeUnit.SECONDS);
}

/**
 * Closes a Netty channel and stops accepting any command.
 *
 * @param channelProxy the specified channel proxy
 */
public void stopChannelProxy(final ChannelProxy channelProxy) {
    channelProxy.setStopped(true);

    final Channel channel = channelProxy.getChannel();
    EventLoop eventLoop = channel.eventLoop();
    eventLoop.schedule(new Runnable() {
        @Override
        public void run() {
            if (channelProxy.hasWaitingRequests()) {
                // cancel all waiting requests belong to this channel
                channelProxy.cancelWaitingRequests();
            }
            // close this unused channel
            channel.close();
        }
    }, Constants.CANCEL_WAITING_REQUEST_DELAY, TimeUnit.SECONDS);

    log.info("Stop a channel proxy from pool. channel proxy: {}", channelProxy);
}

/**
 * Handles an inactive channel and tries to reconnects original remote server
 */
@Override
public void channelInactive(final ChannelHandlerContext context) throws Exception {
    log.info("Client is disconnected from server: {}", context.channel().remoteAddress());

    ChannelProxy channelProxy = ClientChannelManager.getInstance().findChannelProxy(context.channel());
    if (channelProxy == null || channelProxy.isStopped()) {
        log.warn("Fail to find any matching proxy of client channel or this client channel had been stopped.");
        return;
    }

    log.info("Reconnects to remote server after {} seconds.", Constants.RECONNECT_DELAY);

    // delay several seconds to reconnect the original remote server
    EventLoop eventLoop = context.channel().eventLoop();
    eventLoop.schedule(new Runnable() {
        @Override
        public void run() {
            reconnect(context);
        }
    }, Constants.RECONNECT_DELAY, TimeUnit.SECONDS);
}

/**
 * Create a new instance
 *
 * @param parent        the {@link Channel} which is the parent of this {@link NioSctpChannel}
 *                      or {@code null}.
 * @param sctpChannel   the underlying {@link SctpChannel}
 */
public NioSctpChannel(Channel parent, EventLoop eventLoop, SctpChannel sctpChannel) {
    super(parent, eventLoop, sctpChannel, SelectionKey.OP_READ);
    try {
        sctpChannel.configureBlocking(false);
        config = new DefaultSctpChannelConfig(this, sctpChannel);
        notificationHandler = new SctpNotificationHandler(this);
    } catch (IOException e) {
        try {
            sctpChannel.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "Failed to close a partially initialized sctp channel.", e2);
            }
        }

        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

@Override
public ChannelFuture shutdownOutput(final ChannelPromise future) {
    EventLoop loop = eventLoop();
    if (loop.inEventLoop()) {
        try {
            socket.shutdownOutput();
            future.setSuccess();
        } catch (Throwable t) {
            future.setFailure(t);
        }
    } else {
        loop.execute(new Runnable() {
            @Override
            public void run() {
                shutdownOutput(future);
            }
        });
    }
    return future;
}

public static ChannelConfiguration clientConfig(EventLoopGroup workerGroup) {
  EventLoopGroup parent = workerGroup;
  if (parent instanceof EventLoop) {
    parent = ((EventLoop) workerGroup).parent();
  }
  Class<? extends Channel> channelClass;
  if (parent instanceof EpollEventLoopGroup) {
    channelClass = EpollSocketChannel.class;
  } else if (parent instanceof NioEventLoopGroup) {
    channelClass = NioSocketChannel.class;
  } else {
    throw new RuntimeException("Unsupported EventLoopGroup " + workerGroup.getClass());
  }

  return new ChannelConfiguration(workerGroup, channelClass);
}

/**
 * This method will configure a worker EventLoopGroup and a Channel for use by a client. It will
 * try to use the correct SocketChannel for the provided workerGroup.
 *
 * @param workerGroup uses EventLoopGroup in the ClientChannelConfiguration
 * @return ClientChannelConfiguration
 */
public static ClientChannelConfiguration clientConfig(EventLoopGroup workerGroup) {
  EventLoopGroup parent = workerGroup;
  if (parent instanceof EventLoop) {
    parent = ((EventLoop) workerGroup).parent();
  }
  Class<? extends Channel> channelClass;
  if (parent instanceof EpollEventLoopGroup) {
    channelClass = EpollSocketChannel.class;
  } else if (parent instanceof NioEventLoopGroup) {
    channelClass = NioSocketChannel.class;
  } else {
    throw new RuntimeException("Unsupported EventLoopGroup " + workerGroup.getClass());
  }

  return new ClientChannelConfiguration(workerGroup, channelClass);
}

@Override
public void operationComplete(final ChannelFuture future) throws Exception {
    if (future.isCancelled()) {
        LOG.debug("Connection {} cancelled!", future);
    } else if (future.isSuccess()) {
        LOG.debug("Connection {} succeeded!", future);
        future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> scheduleConnect());
    } else {
        if (this.delay > MAXIMUM_BACKOFF) {
            LOG.warn("The time of maximum backoff has been exceeded. No further connection attempts with BMP "
                    + "router {}.", this.remoteAddress);
            future.cancel(false);
            return;
        }
        final EventLoop loop = future.channel().eventLoop();
        loop.schedule(() -> this.bootstrap.connect().addListener(this), this.delay, TimeUnit.MILLISECONDS);
        LOG.info("The connection try to BMP router {} failed. Next reconnection attempt in {} milliseconds.",
                this.remoteAddress, this.delay);
        this.delay *= 2;
    }
}

synchronized void reconnect() {
    if (this.retryTimer == 0) {
        LOG.debug("Retry timer value is 0. Reconnection will not be attempted");
        this.setFailure(this.pending.cause());
        return;
    }

    final EventLoop loop = this.pending.channel().eventLoop();
    loop.schedule(() -> {
        synchronized (BGPProtocolSessionPromise.this) {
            if (BGPProtocolSessionPromise.this.peerSessionPresent) {
                LOG.debug("Connection to {} already exists", BGPProtocolSessionPromise.this.address);
                BGPProtocolSessionPromise.this.connectSkipped = true;
                return;
            }

            BGPProtocolSessionPromise.this.connectSkipped = false;
            LOG.debug("Attempting to connect to {}", BGPProtocolSessionPromise.this.address);
            final ChannelFuture reconnectFuture = BGPProtocolSessionPromise.this.bootstrap.connect();
            reconnectFuture.addListener(new BootstrapConnectListener());
            BGPProtocolSessionPromise.this.pending = reconnectFuture;
        }
    }, this.retryTimer, TimeUnit.SECONDS);
    LOG.debug("Next reconnection attempt in {}s", this.retryTimer);
}

@Override
public void operationComplete(ChannelFuture future) throws Exception {
    if (future.isSuccess()) {
        log.info(String.format(MSG_STATE,
                ofSwitch.dpid(),
                MSG_CONNECTED,
                controller.ip(),
                controller.port()));
        // FIXME add close future listener to handle connection lost
    } else {
        if (retryCount.getAndIncrement() > MAX_RETRY) {
            log.warn(String.format(MSG_STATE,
                    ofSwitch.dpid(),
                    MSG_FAILED,
                    controller.ip(),
                    controller.port()));
        } else {
            final EventLoop loop = future.channel().eventLoop();
            loop.schedule(this::connect, 1L, TimeUnit.SECONDS);
        }
    }
}

/**
 * @see {@link #connect()}
 */
private ChannelFuture doConnect(final SocketAddress remoteAddress,
        final SocketAddress localAddress, EventLoop eventLoop) {
    final ChannelFuture regFuture = initAndRegister(eventLoop);
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    final ChannelPromise promise = channel.newPromise();
    if (regFuture.isDone()) {
        doConnect0(regFuture, channel, remoteAddress, localAddress, promise);
    } else {
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                doConnect0(regFuture, channel, remoteAddress, localAddress, promise);
            }
        });
    }
    return promise;
}

public static WorkerGroup newInstance(SSConnection ssCon, StorageGroup sg, PersistentDatabase ctxDB) throws PEException {
    WorkerGroup wg = workerGroupPool.get(sg, ssCon.getUserAuthentication());
          Channel channel = ssCon.getChannel();
          EventLoop eventLoop = channel == null ? null : channel.eventLoop();

    if (wg == null) {
        wg = new WorkerGroup(sg).provision(ssCon, ssCon, ssCon.getUserAuthentication(), eventLoop);
    } else {
              wg.bindToClientThread(eventLoop);
          }
    try {
        if (ctxDB != null) 
            wg.setDatabase(ssCon, ctxDB);
        wg.assureSessionVariables(ssCon);
    } catch (PEException e) {
              if (logger.isDebugEnabled())
            logger.debug("NPE: WorkerGroupFactory.newInstance() calls releaseWorkers() on "+ wg);
        wg.releaseWorkers(ssCon);
        throw e;
    }
    if (wg.workerMap == null)
        throw new PECodingException("WorkerGroupFactory.newInstance() returns previously closed worker group");
    return wg;
}

@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
    if (!channelFuture.isSuccess()) {
        channelFuture.channel().close();

        if (count.incrementAndGet() < MAX_RETRY) {
            final EventLoop loop = channelFuture.channel().eventLoop();

            loop.schedule(() -> {
                try {
                    controller.connectRetry(this.ip, this.port, this);
                } catch (Exception e) {
                    log.warn("Connection to the ovsdb server {}:{} failed(cause: {})", ip, port, e);
                }
            }, 1L, TimeUnit.SECONDS);
        } else {
            failhandler.accept(new Exception("max connection retry(" + MAX_RETRY + ") exceeded"));
        }
    } else {
        handleNewNodeConnection(channelFuture.channel());
    }
}

@Override
@SuppressWarnings("unchecked")
public T newChannel(EventLoop eventLoop, EventLoopGroup childGroup) {
    switch (kind()) {
    case ACCEPTOR:
        switch (type()) {
        case DATAGRAM:
            return (T) new NioUdtMessageAcceptorChannel(eventLoop, childGroup);
        case STREAM:
            return (T) new NioUdtByteAcceptorChannel(eventLoop, childGroup);
        default:
            throw new IllegalStateException("wrong type: " + type());
        }
    case CONNECTOR:
    case RENDEZVOUS:
    default:
        throw new IllegalStateException("wrong kind: " + kind());
    }
}

protected NioUdtAcceptorChannel(EventLoop eventLoop, EventLoopGroup childGroup,
        ServerSocketChannelUDT channelUDT) {
    super(null, eventLoop, childGroup, channelUDT, OP_ACCEPT);
    try {
        channelUDT.configureBlocking(false);
        config = new DefaultUdtServerChannelConfig(this, channelUDT, true);
    } catch (final Exception e) {
        try {
            channelUDT.close();
        } catch (final Exception e2) {
            if (logger.isWarnEnabled()) {
                logger.warn("Failed to close channel.", e2);
            }
        }
        throw new ChannelException("Failed to configure channel.", e);
    }
}

/**
 * verify factory
 */
@Test
public void provideFactory() {

    EventLoop loop = new NioEventLoopGroup().next();
    EventLoopGroup childGroup = new NioEventLoopGroup();

    // bytes
    assertNotNull(NioUdtProvider.BYTE_ACCEPTOR.newChannel(loop, childGroup));
    assertNotNull(NioUdtProvider.BYTE_CONNECTOR.newChannel(loop));
    assertNotNull(NioUdtProvider.BYTE_RENDEZVOUS.newChannel(loop));

    // message
    assertNotNull(NioUdtProvider.MESSAGE_ACCEPTOR.newChannel(loop, childGroup));
    assertNotNull(NioUdtProvider.MESSAGE_CONNECTOR.newChannel(loop));
    assertNotNull(NioUdtProvider.MESSAGE_RENDEZVOUS.newChannel(loop));

    // acceptor types
    assertTrue(NioUdtProvider.BYTE_ACCEPTOR.newChannel(loop, childGroup) instanceof UdtServerChannel);
    assertTrue(NioUdtProvider.MESSAGE_ACCEPTOR.newChannel(loop, childGroup) instanceof UdtServerChannel);
}

@Override
public ChannelFuture shutdownOutput(final ChannelPromise future) {
    EventLoop loop = eventLoop();
    if (loop.inEventLoop()) {
        try {
            socket.shutdownOutput();
            future.setSuccess();
        } catch (Throwable t) {
            future.setFailure(t);
        }
    } else {
        loop.execute(new Runnable() {
            @Override
            public void run() {
                shutdownOutput(future);
            }
        });
    }
    return future;
}