(like proxy service) sets a connect future listener that does writes.
435020
@Override
    public void doSessionCreated(ManagementContext managementContext,
                                 ServiceManagementBean serviceBean,
                                 IoSessionEx session,
                                 ManagementSessionType managementSessionType) throws Exception {
        SessionManagementBean sessionBean = managementContext.addSessionManagementBean(serviceBean, session);
        serviceBean.storeSessionManagementBean(sessionBean);

        managementContext.getManagementSessionStrategy().doSessionCreated(sessionBean);

        super.doSessionCreated(managementContext, serviceBean, session, managementSessionType);
    }
291363
void becomeConnected(Sink sink, Socket socket) {
    checkState(this.sink == null, "AsyncSink's becomeConnected should only be called once.");
    this.sink = checkNotNull(sink, "sink");
    this.socket = checkNotNull(socket, "socket");
  }
956806
public void doReconnect() {
        if (channelRef.get() != null) {
            // if (channelRef.get().isWritable()) {
            // LOG.info("already exist a writable channel, give up reconnect, {}",
            // channelRef.get());
            // return;
            // }
            return;
        }

        if (isClosed()) {
            return;
        }

        if (isConnecting.getAndSet(true)) {
            LOG.info("Connect twice {}", name());
            return;
        }

        long sleepMs = getSleepTimeMs();
        LOG.info("Reconnect ... [{}], {}, sleep {}ms", retries.get(), name, sleepMs);
        ChannelFuture future = bootstrap.connect(remoteAddr);
        future.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                isConnecting.set(false);
                Channel channel = future.getChannel();
                if (future.isSuccess()) {
                    // do something else
                    LOG.info("Connection established, channel = :{}", channel);
                    setChannel(channel);
                    // handleResponse();
                    BATCH_THRESHOLD_WARN = ConfigExtension.getNettyBufferThresholdSize(stormConf);

                    // Check if any pending message
                    /*synchronized (writeLock) {
                        if (channel != null && messageBuffer.size() > 0) {
                            MessageBatch messageBatch = messageBuffer.drain();
                            flushRequest(channel, messageBatch);
                        } else {
                            LOG.warn("Failed to flush pending message after reconnecting, channel={}, messageBuffer.size={}",
                                    channel, messageBuffer.size());
                        }
                    }*/
                } else {
                    if (!isClosed()) {
                        LOG.info("Failed to reconnect ... [{}], {}, channel = {}, cause = {}",
                                retries.get(), name, channel, future.getCause());
                        reconnect();
                    }
                }
            }
        });
        JStormUtils.sleepMs(sleepMs);
    }
627324
public void connect(Context context, BeanListener listener) {
        lastKnownContext = context;
        beanListener = listener;
        gattClient.connect(context, device);
    }
167095
private void createProxy(final Class<L> listenerInterface, final ClassLoader classLoader) {
        proxy = listenerInterface.cast(Proxy.newProxyInstance(classLoader,
                new Class[] { listenerInterface }, createInvocationHandler()));
    }
855260
private void notifyConnectMessage(HttpMessage connectMessage) {
        for (ConnectRequestProxyListener listener : parentServer.getConnectRequestProxyListeners()) {
            try {
                listener.receivedConnectRequest(connectMessage);
            } catch (Exception e) {
                log.error("An error occurred while notifying listener:", e);
            }
        }
    }
1204988
private void scheduleReconnection() {

        if (runnable != null) {
            handler.removeCallbacks(runnable);
        }

        runnable = () -> {
            if (shouldReconnect() && (socket == null || !socket.isOpen()) && retryStrategy.retry() && !isNetworkUnavailable) {
                log.d("Reconnecting socket");
                connect();
            }
        };
        long delay = retryStrategy.getDelay();
        handler.postDelayed(runnable, delay);
        log.d("Socket reconnection in " + delay / 1000 + " seconds.");
    }
1101966
private CompletableFuture<Void> listen() {
    CompletableFuture<Void> future = new CompletableFuture<>();
    context.getThreadContext().executor().execute(() -> {
      internalServer.listen(cluster().member().serverAddress(), context::connectServer).whenComplete((internalResult, internalError) -> {
        if (internalError == null) {
          // If the client address is different than the server address, start a separate client server.
          if (clientServer != null) {
            clientServer.listen(cluster().member().clientAddress(), context::connectClient).whenComplete((clientResult, clientError) -> {
              started = true;
              future.complete(null);
            });
          } else {
            started = true;
            future.complete(null);
          }
        } else {
          future.completeExceptionally(internalError);
        }
      });
    });

    return future;
  }
1049042
private static <V> ListenableFuture<V> closedChannelFuture() {
    SettableFuture<V> future = SettableFuture.create();
    future.setException(new ClosedChannelException());
    return future;
  }
1052637
def connect(self, callback=None, timeout=None):
        '''
        
        '''
        if hasattr(self, '_connecting_future') and not self._connecting_future.done():  # pylint: disable=E0203
            future = self._connecting_future  # pylint: disable=E0203
        else:
            if hasattr(self, '_connecting_future'):
                # read previous future result to prevent the "unhandled future exception" error
                self._connecting_future.exception()  # pylint: disable=E0203
            future = tornado.concurrent.Future()
            self._connecting_future = future
            self._connect(timeout=timeout)

        if callback is not None:
            def handle_future(future):
                response = future.result()
                self.io_loop.add_callback(callback, response)
            future.add_done_callback(handle_future)

        return future
262464
function Connection(endpoint, options, connectionListener) {
  if (!(this instanceof Connection)) return new Connection(endpoint, options, connectionListener);

  stream.Duplex.call(this);

  if (endpoint && !(endpoint instanceof Endpoint)) {
    endpoint = new Endpoint(endpoint);
  }
  if (typeof options === 'function') {
    connectionListener = options;
    options = undefined;
  }
  this.endpoint = endpoint;
  this.options = options || {};
  if (typeof connectionListener === 'function') {
    this.on('connect', connectionListener);
  }
  this.status = {sent: 0, received: 0};

  this.on('_read_data', function (data) {
    this.push(data);
    this.status.received += data.length;
  });

  if (endpoint) {
    //connect
    connect.call(this);
  }
}
1349516
@SuppressWarnings("unchecked")
  private void setupConnection(Address address, Connection connection, CompletableFuture<Connection> future) {
    LOGGER.debug("{} - Setting up connection to {}", id, address);

    this.connection = connection;

    connection.onClose(c -> {
      if (c.equals(this.connection)) {
        LOGGER.debug("{} - Connection closed", id);
        this.connection = null;
      }
    });
    connection.onException(c -> {
      if (c.equals(this.connection)) {
        LOGGER.debug("{} - Connection lost", id);
        this.connection = null;
      }
    });

    for (Map.Entry<Class<?>, Function> entry : handlers.entrySet()) {
      connection.handler(entry.getKey(), entry.getValue());
    }

    // When we first connect to a new server, first send a ConnectRequest to the server to establish
    // the connection with the server-side state machine.
    ConnectRequest request = ConnectRequest.builder()
      .withClientId(id)
      .build();

    LOGGER.trace("{} - Sending {}", id, request);
    connection.<ConnectRequest, ConnectResponse>sendAndReceive(request).whenComplete((r, e) -> handleConnectResponse(r, e, future));
  }
1164760
def on_connect(self, connection):
        ""
        # NOTE: for python3, we can't pass bytestrings as keyword arguments
        # so we need to decode channel/pattern names back to unicode strings
        # before passing them to [p]subscribe.
        self.pending_unsubscribe_channels.clear()
        self.pending_unsubscribe_patterns.clear()
        if self.channels:
            channels = {}
            for k, v in iteritems(self.channels):
                channels[self.encoder.decode(k, force=True)] = v
            self.subscribe(**channels)
        if self.patterns:
            patterns = {}
            for k, v in iteritems(self.patterns):
                patterns[self.encoder.decode(k, force=True)] = v
            self.psubscribe(**patterns)
1259886
void notifyListenersNetworkUneachable() {
        // NOTE: synchronized(listener) causes deadlock with listeners and Replicator.lock.
        final Set<NetworkReachabilityListener> copy = new HashSet<>(listeners);
        for (NetworkReachabilityListener listener : copy) {
            if (listener != null) { listener.networkUnreachable(); }
        }
    }
586862
private void setReadyForRead() throws ProtocolException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "setReadyForRead entry: stream id:" + myID);
        }
        muxLink.updateHighestStreamId(myID);
        if (headersCompleted) {
            ExecutorService executorService = CHFWBundle.getExecutorService();
            Http2Ready readyThread = new Http2Ready(h2HttpInboundLinkWrap);
            executorService.execute(readyThread);
        }
    }
1215401
public void connectedReceived() {
        if (null != listenerReference) {
            final Listener listener = listenerReference.get();
            if (null != listener) {
                new Handler(Looper.getMainLooper()).post(new Runnable() {
                    public void run() {
                        if (isActive()) {
                            listener.onConnected(RespokeCall.this);
                        }
                    }
                });
            }
        }
    }
468904
public void addFuture(ListenableFuture<Boolean, OperationCompletionListener> future) {
        future.addListener(this);
        if (future.isDone()) countDown();
        this.futures.add(future);
    }