connect() — netty Function Reference

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringChannel.java lines 1055–1166

        @Override
        public void connect(
                final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
            // Don't mark the connect promise as uncancellable as in fact we can cancel it as it is using
            // non-blocking io.
            if (promise.isDone() || !ensureOpen(promise)) {
                return;
            }

            if (delayedClose != null) {
                promise.tryFailure(annotateConnectException(new ClosedChannelException(), remoteAddress));
                return;
            }
            try {
                if (connectPromise != null) {
                    throw new ConnectionPendingException();
                }
                if (localAddress instanceof InetSocketAddress) {
                    checkResolvable((InetSocketAddress) localAddress);
                }

                if (remoteAddress instanceof InetSocketAddress) {
                    checkResolvable((InetSocketAddress) remoteAddress);
                }

                if (remote != null) {
                    // Check if already connected before trying to connect. This is needed as connect(...) will not#
                    // return -1 and set errno to EISCONN if a previous connect(...) attempt was setting errno to
                    // EINPROGRESS and finished later.
                    throw new AlreadyConnectedException();
                }

                if (localAddress != null) {
                    socket.bind(localAddress);
                }

                if (remoteAddress instanceof InetSocketAddress) {
                    InetSocketAddress inetSocketAddress = (InetSocketAddress) remoteAddress;
                    ByteBuf initialData = null;
                    if (IoUring.isTcpFastOpenClientSideAvailable() &&
                        config().getOption(ChannelOption.TCP_FASTOPEN_CONNECT) == Boolean.TRUE) {
                        ChannelOutboundBuffer outbound = unsafe().outboundBuffer();
                        outbound.addFlush();
                        Object curr;
                        if ((curr = outbound.current()) instanceof ByteBuf) {
                            initialData = (ByteBuf) curr;
                        }
                    }
                    if (initialData != null) {
                        msgHdrMemoryArray = new MsgHdrMemoryArray((short) 1);
                        MsgHdrMemory hdr = msgHdrMemoryArray.hdr(0);
                        hdr.set(socket, inetSocketAddress, IoUring.memoryAddress(initialData),
                                initialData.readableBytes(), (short) 0);

                        int fd = fd().intValue();
                        IoRegistration registration = registration();
                        IoUringIoOps ops = IoUringIoOps.newSendmsg(fd, (byte) 0, Native.MSG_FASTOPEN,
                                hdr.address(), hdr.idx());
                        connectId = registration.submit(ops);
                        if (connectId == 0) {
                            // Directly release the memory if submitting failed.
                            freeMsgHdrArray();
                        }
                    } else {
                        submitConnect(inetSocketAddress);
                    }
                } else if (remoteAddress instanceof DomainSocketAddress) {
                    DomainSocketAddress unixDomainSocketAddress = (DomainSocketAddress) remoteAddress;
                    submitConnect(unixDomainSocketAddress);
                } else {
                    throw new Error("Unexpected SocketAddress implementation " + className(remoteAddress));
                }

                if (connectId != 0) {
                    ioState |= CONNECT_SCHEDULED;
                }
            } catch (Throwable t) {
                closeIfClosed();
                promise.tryFailure(annotateConnectException(t, remoteAddress));
                return;
            }