public boolean reportEvent(SendQos sendQos, TcpNettyClient client,
            EncodeObject encObject, MsgSendCallback callback, ProcessResult procResult) {
        if (!this.started.get()) {
            return procResult.setFailResult(ErrorCode.SDK_CLOSED);
        }
        long clientTerm = client.getChanTermId();
        if (sendQos == SendQos.NO_ACK) {
            // process no ack report
            if (client.write(clientTerm, encObject, procResult)) {
                client.decInFlightMsgCnt(clientTerm);
            }
            return procResult.isSuccess();
        }
        TcpCallFuture newFuture = new TcpCallFuture(encObject,
                client.getClientAddr(), clientTerm, client.getChanStr(), callback);
        TcpCallFuture curFuture = reqObjects.putIfAbsent(encObject.getMessageId(), newFuture);
        if (curFuture != null) {
            if (sendExceptCnt.shouldPrint()) {
                logger.warn("ClientMgr({}) found message id {} has existed.",
                        senderId, encObject.getMessageId());
            }
            return procResult.setFailResult(ErrorCode.DUPLICATED_MESSAGE_ID);
        }
        // build channel -- messageId map
        ConcurrentHashMap<Integer, Boolean> msgIdMap = channelMsgIdMap.get(client.getChanStr());
        if (msgIdMap == null) {
            ConcurrentHashMap<Integer, Boolean> tmpMsgIdMap = new ConcurrentHashMap<>();
            msgIdMap = channelMsgIdMap.putIfAbsent(client.getChanStr(), tmpMsgIdMap);
            if (msgIdMap == null) {
                msgIdMap = tmpMsgIdMap;
            }
        }
        msgIdMap.put(encObject.getMessageId(), newFuture.isAsyncCall());
        // send message
        if (newFuture.isAsyncCall()) {
            // process async report
            reqTimeouts.put(encObject.getMessageId(),
                    timerObj.newTimeout(new TimeoutTask(encObject.getMessageId()),
                            tcpConfig.getRequestTimeoutMs(), TimeUnit.MILLISECONDS));
            if (!client.write(clientTerm, encObject, procResult)) {
                Timeout timeout = reqTimeouts.remove(newFuture.getMessageId());
                if (timeout != null) {
                    timeout.cancel();
                }
                rmvMsgStubInfo(newFuture.getMessageId());
            }
            return procResult.isSuccess();
        } else {
            // process sync report
            if (!client.write(clientTerm, encObject, procResult)) {
                rmvMsgStubInfo(newFuture.getMessageId());
                return false;
            }
            boolean retValue = newFuture.get(procResult,
                    tcpConfig.getRequestTimeoutMs(), TimeUnit.MILLISECONDS);
            if (rmvMsgStubInfo(newFuture.getMessageId())) {
                if (procResult.getErrCode() == ErrorCode.SEND_WAIT_TIMEOUT.getErrCode()) {
                    client.setBusy(clientTerm);
                }
                client.decInFlightMsgCnt(clientTerm);
            }
            return retValue;
        }
    }
Build up the connection between the server and client.
inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/network/tcp/TcpClientMgr.java