Method NehServiceQueryCallback

library/cpp/par/par_network.cpp:92–172

        }

        void NehServiceQueryCallback(const NNeh::IRequestRef& req) {
            CHROMIUM_TRACE_FUNCTION();

            PAR_DEBUG_LOG << "At " << GetHostAndPort() << " incoming req: " << req->Scheme() << " " << req->RemoteHost() << " " << req->Service() << Endl;
            if (req->Canceled()) {
                PAR_DEBUG_LOG << "At " << GetHostAndPort() << " incoming req: " << req->Scheme() << " " << req->RemoteHost() << " " << req->Service() << " query is canceled in flight, don't even try to launch it" << Endl;
                // if query is canceled in flight, don't even try to launch it
                return;
            }
            auto del1pos = req->Data().find('\xff');
            auto del2pos = req->Data().find('\xff', del1pos + 1);
            if (del1pos == TStringBuf::npos || del2pos == TStringBuf::npos) {
                req->SendError(NNeh::IRequest::BadRequest, "no \\xff delimiters found");
                return;
            }

            TGUID reqId;
            if (!GetGuid(ToString(req->Data().substr(0, del1pos)), reqId)) {
                req->SendError(NNeh::IRequest::BadRequest, "incorrect guid");
                return;
            }
            auto Url = req->Data().substr(del1pos + 1, del2pos - del1pos - 1);
            auto del3pos = req->Data().find('\xff', del2pos + 1);
            TVector<char> Data;
            if (del3pos != TStringBuf::npos) {
                auto crc32ref = FromString<ui32>(TStringBuf(req->Data().begin() + del2pos + 1, del3pos - del2pos - 1));
                auto dataSize = req->Data().end() - (req->Data().begin() + del3pos + 1);
                auto crc32actual = Crc32c(req->Data().begin() + del3pos + 1, dataSize);
                if (crc32actual != crc32ref) {
                    TStringBuilder errorString;
                    errorString << "Invalid crc32 for data, expected " << crc32ref << " got " << crc32actual;

                    PAR_DEBUG_LOG << "At " << GetHostAndPort() << ": " << errorString << Endl;
                    req->SendError(NNeh::IRequest::BadRequest, errorString);
                    return;
                }
                Data.assign(req->Data().begin() + del3pos + 1, req->Data().end());
            }
            QuickLZDecompress(&Data);
            PAR_DEBUG_LOG << "At " << GetHostAndPort() << " got request " << GetGuidAsString(reqId) << " service: " << Url << " data len: " << Data.size() << Endl;
            NNeh::TData ok = {'O', 'K'};
            req->SendReply(ok);
            if (Url == "_ping_") {
                return;
            } else if (Url == "_cancel_") {
                if (!IncomingRequestsInfo.EraseValueIfPresent(reqId)) {
                    return;
                }
                QueryCancelCallback(reqId);
            } else if (Url == "_reply_") {
                if (!RequestsInfo.EraseValueIfPresent(reqId)) {
                    return;
                }
                TAutoPtr<TNetworkResponse> httpResponse = new TNetworkResponse;
                httpResponse->ReqId = reqId;
                httpResponse->Data = std::move(Data);
                httpResponse->Status = TNetworkResponse::EStatus::Ok;
                auto directReplyNotifier = [&httpResponse](TIntrusivePtr<TSyncRequestsInfo>& syncRequestInfo) {