reader_id1.cpp

来自「ncbi源码」· C++ 代码 · 共 1,248 行 · 第 1/3 页

                     exc.what());
        }
    }
    m_NoMoreConnections = true;
    return 0;
}


int CId1Reader::ResolveSeq_id_to_gi(const CSeq_id& seqId, TConn conn)
{
    CID1server_request id1_request;
    id1_request.SetGetgi(const_cast<CSeq_id&>(seqId));

    CID1server_back id1_reply;
    x_ResolveId(id1_reply, id1_request, conn);

    return id1_reply.IsGotgi()? id1_reply.GetGotgi(): 0;
}


typedef pair<const char*, CReader::ESatellite> TSatPair;
static const TSatPair sc_SatArray[] = {
    TSatPair("SNP",        CReader::eSatellite_SNP),
    TSatPair("ti",         CReader::eSatellite_TRACE),
    TSatPair("TR_ASSM_CH", CReader::eSatellite_TR_ASSM_CH),
    TSatPair("TRACE_ASSM", CReader::eSatellite_TRACE_ASSM),
    TSatPair("TRACE_CHGR", CReader::eSatellite_TRACE_CHGR)
};
typedef CStaticArrayMap<const char*, CReader::ESatellite, PNocase> TSatMap;
static const TSatMap sc_SatMap(sc_SatArray, sizeof (sc_SatArray));


void CId1Reader::ResolveSeq_id(TSeqrefs& srs, const CSeq_id& id, TConn conn)
{
    if ( id.IsGeneral()  &&  id.GetGeneral().GetTag().IsId() ) {
        const CDbtag& dbtag = id.GetGeneral();
        const CObject_id& objid = dbtag.GetTag();

        int sat = 0;
        int sat_key = objid.GetId();
        if (sat_key != 0) {
            TSatMap::const_iterator iter =
                sc_SatMap.find(dbtag.GetDb().c_str());
            if (iter != sc_SatMap.end()) {
                sat = iter->second;
            } else {
                try {
                    sat = NStr::StringToInt(dbtag.GetDb());
                }
                catch (...) {
                }
            }

            if (sat != 0) {
                srs.push_back(Ref(new CSeqref(0, sat, sat_key)));
                return;
            }
        }
    }
    CReader::ResolveSeq_id(srs, id, conn);
}


void CId1Reader::RetrieveSeqrefs(TSeqrefs& srs, int gi, TConn conn)
{
    CID1server_request id1_request;
    {{
        CID1server_maxcomplex& blob = id1_request.SetGetblobinfo();
        blob.SetMaxplex(eEntry_complexities_entry);
        blob.SetGi(gi);
    }}
    
    CID1server_back id1_reply;
    x_ResolveId(id1_reply, id1_request, conn);

    if ( !id1_reply.IsGotblobinfo() ) {
        return;
    }

    const CID1blob_info& info = id1_reply.GetGotblobinfo();
    if ( info.GetWithdrawn() > 0 || info.GetConfidential() > 0 ) {
        LOG_POST(Warning<<"CId1Reader::RetrieveSeqrefs("<<gi<<"): "
                 "gi is private");
        return;
    }
    if ( info.GetSat() < 0 || info.GetSat_key() < 0 ) {
        LOG_POST(Warning<<"CId1Reader::RetrieveSeqrefs("<<gi<<"): "
                 "negative sat/satkey");
        return;
    }
    CRef<CSeqref> ref(new CSeqref(gi, info.GetSat(), info.GetSat_key()));
    ref->SetVersion(x_GetVersion(info));
    srs.push_back(ref);
   
    if ( TrySNPSplit() ) {
        if ( !info.IsSetExtfeatmask() ) {
            AddSNPSeqref(srs, gi, CSeqref::fPossible);
        }
        else if ( info.GetExtfeatmask() & 1 ) {
            AddSNPSeqref(srs, gi);
        }
    }
}


int CId1Reader::GetVersion(const CSeqref& seqref, TConn conn)
{
    if ( seqref.GetVersion() == 0 ) {
        const_cast<CSeqref&>(seqref).SetVersion(x_GetVersion(seqref, conn));
    }
    return seqref.GetVersion();
}


int CId1Reader::x_GetVersion(const CSeqref& seqref, TConn conn)
{
    CID1server_request id1_request;
    x_SetParams(seqref, id1_request.SetGetblobinfo(), IsSNPSeqref(seqref));
    
    CID1server_back    id1_reply;
    x_ResolveId(id1_reply, id1_request, conn);
    
    if ( id1_reply.IsGotblobinfo() ) {
        return x_GetVersion(id1_reply.GetGotblobinfo());
    }
    else {
        return 1; // default non-zero version
    }
}


void CId1Reader::x_ResolveId(CID1server_back& id1_reply,
                             const CID1server_request& id1_request,
                             TConn conn)
{
#ifdef ID1_COLLECT_STATS
    CStopWatch sw;
    if ( CollectStatistics() ) {
        sw.Start();
    }
#endif

    CConn_ServiceStream* stream = x_GetConnection(conn);
    {{
        CObjectOStreamAsnBinary out(*stream);
        out << id1_request;
        out.Flush();
    }}
    
    {{
        CObjectIStreamAsnBinary in(*stream);
        in >> id1_reply;
    }}
    /*
    if ( id1_reply.IsError() && id1_reply.GetError() == 0 ) {
        char next_byte;
        if ( CStreamUtils::Readsome(*stream, &next_byte, 1) ) {
            CStreamUtils::Pushback(*stream, &next_byte, 1);
            ERR_POST("Extra reply from ID1 server: ERROR 0");
            CObjectIStreamAsnBinary in(*stream);
            in >> id1_reply;
        }
    }
    */

#ifdef ID1_COLLECT_STATS
    if ( CollectStatistics() ) {
        double time = sw.Elapsed();
        if ( id1_request.Which() == CID1server_request::e_Getgi ) {
            LogStat("CId1Reader: resolved id", id1_request.GetGetgi(), time);
            resolve_id_count++;
            resolve_id_time += time;
        }
        else if ( id1_request.Which() == CID1server_request::e_Getblobinfo ) {
            const CID1server_maxcomplex& req = id1_request.GetGetblobinfo();
            if ( req.IsSetSat() ) {
                LogStat("CId1Reader: got blob version", req, time);
                resolve_ver_count++;
                resolve_ver_time += time;
            }
            else {
                LogStat("CId1Reader: resolved gi", req.GetGi(), time);
                resolve_gi_count++;
                resolve_gi_time += time;
            }
        }
    }
#endif
}


CRef<CTSE_Info> CId1Reader::GetTSEBlob(const CSeqref& seqref,
                                       TConn conn)
{
    if ( seqref.GetFlags() & CSeqref::fPrivate ) {
        NCBI_THROW(CLoaderException, ePrivateData, "gi is private");
    }
    CID1server_back id1_reply;
    CRef<CID2S_Split_Info> split_info;
    x_GetTSEBlob(id1_reply, split_info, seqref, conn);

    CRef<CSeq_entry> seq_entry;
    bool dead = false;
    switch ( id1_reply.Which() ) {
    case CID1server_back::e_Gotseqentry:
        seq_entry.Reset(&id1_reply.SetGotseqentry());
        break;
    case CID1server_back::e_Gotdeadseqentry:
        dead = true;
        seq_entry.Reset(&id1_reply.SetGotdeadseqentry());
        break;
    case CID1server_back::e_Error:
        switch ( id1_reply.GetError() ) {
        case 1:
            NCBI_THROW(CLoaderException, ePrivateData, "id is withdrawn");
        case 2:
            NCBI_THROW(CLoaderException, ePrivateData, "id is private");
        case 10:
            NCBI_THROW(CLoaderException, eNoData, "invalid args");
        }
        ERR_POST("CId1Reader::GetMainBlob: ID1server-back.error "<<
                 id1_reply.GetError());
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "ID1server-back.error");
        break;
    default:
        // no data
        NCBI_THROW(CLoaderException, eLoaderFailed,
                   "bad ID1server-back type");
    }
    CRef<CTSE_Info> ret(new CTSE_Info(*seq_entry, dead));
    if ( split_info ) {
        CSplitParser::Attach(*ret, *split_info);
    }
    return ret;
}


CRef<CSeq_annot_SNP_Info> CId1Reader::GetSNPAnnot(const CSeqref& seqref,
                                                  TConn conn)
{
    CRef<CSeq_annot_SNP_Info> ret(new CSeq_annot_SNP_Info);

    x_GetSNPAnnot(*ret, seqref, conn);

    return ret;
}


void Id1ReaderSkipBytes(CByteSourceReader& reader, size_t to_skip)
{
    // skip 2 bytes of hacked header
    const size_t kBufferSize = 128;
    char buffer[kBufferSize];
    while ( to_skip ) {
        size_t cnt = reader.Read(buffer, min(to_skip, sizeof(buffer)));
        if ( cnt == 0 ) {
            NCBI_THROW(CEofException, eEof,
                       "unexpected EOF while skipping ID1 SNP wrapper bytes");
        }
        to_skip -= cnt;
    }
}


void CId1Reader::x_GetTSEBlob(CID1server_back& id1_reply,
                              CRef<CID2S_Split_Info>& /*split_info*/,
                              const CSeqref& seqref,
                              TConn conn)
{
#ifdef ID1_COLLECT_STATS
    CStopWatch sw;
    if ( CollectStatistics() ) {
        sw.Start();
    }
    
    try {
#endif
        CConn_ServiceStream* stream = x_GetConnection(conn);
        x_SendRequest(seqref, stream, false);
        x_ReadTSEBlob(id1_reply, seqref, *stream);
        
#ifdef ID1_COLLECT_STATS
        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogBlobStat("CId1Reader: read blob",
                        seqref, last_object_bytes, time);
            main_blob_count++;
            main_bytes += last_object_bytes;
            main_time += time;
        }
    }
    catch ( ... ) {
        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogBlobStat("CId1Reader: read fail blob",
                        seqref, 0, time);
            main_blob_count++;
            main_time += time;
        }
        throw;
    }
#endif
}


void CId1Reader::x_GetSNPAnnot(CSeq_annot_SNP_Info& snp_info,
                               const CSeqref& seqref,
                               TConn conn)
{
#ifdef ID1_COLLECT_STATS
    CStopWatch sw;

    if ( CollectStatistics() ) {
        sw.Start();
    }

    try {
#endif
        CConn_ServiceStream* stream = x_GetConnection(conn);
        x_SendRequest(seqref, stream, true);

#ifdef ID1_COLLECT_STATS
        size_t compressed;
        double decompression_time;
        double total_read_time;
#endif
        {{
            const size_t kSkipHeader = 2, kSkipFooter = 2;
        
            CStreamByteSourceReader src(0, stream);
        
            Id1ReaderSkipBytes(src, kSkipHeader);
        
            CNlmZipBtRdr src2(&src);
        
            x_ReadSNPAnnot(snp_info, seqref, src2);

#ifdef ID1_COLLECT_STATS
            compressed = src2.GetCompressedSize();
            decompression_time = src2.GetDecompressionTime();
            total_read_time = src2.GetTotalReadTime();
#endif
        
            Id1ReaderSkipBytes(src, kSkipFooter);
        }}

#ifdef ID1_COLLECT_STATS
        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogBlobStat("CId1Reader: read SNP blob",
                        seqref, compressed, time);
            snp_blob_count++;
            snp_compressed += compressed;
            snp_uncompressed += last_object_bytes;
            snp_time += time;
            snp_decompression_time += decompression_time;
            snp_total_read_time += total_read_time;
        }
    }
    catch ( ... ) {
        if ( CollectStatistics() ) {
            double time = sw.Elapsed();
            LogBlobStat("CId1Reader: read fail SNP blob",
                        seqref, 0, time);
            snp_blob_count++;
            snp_time += time;
        }
        throw;
    }
#endif
}


void CId1Reader::x_SetParams(const CSeqref& seqref,
                             CID1server_maxcomplex& params,
                             bool is_snp)
{
    bool is_external = is_snp;
    bool skip_extfeat = !is_external && TrySNPSplit();
    enum {
        kNoExtFeat = 1<<4
    };
    EEntry_complexities maxplex = eEntry_complexities_entry;
    if ( skip_extfeat ) {
        maxplex = EEntry_complexities(int(maxplex) | kNoExtFeat);
    }
    params.SetMaxplex(maxplex);
    params.SetGi(seqref.GetGi());
    params.SetEnt(seqref.GetSatKey());
    params.SetSat(NStr::IntToString(seqref.GetSat()));
}


int CId1Reader::x_GetVersion(const CID1blob_info& info) const
{
    int version = info.GetBlob_state();
    if ( version < 0 ) {
        version = -version;
    }
    if ( version == 0 ) {
        // set to default: 1
        // so that we will not reask version in future
        version = 1;
    }
    return version;
}


void CId1Reader::x_SendRequest(const CSeqref& seqref,
                               CConn_ServiceStream* stream,
                               bool is_snp)
{
    CID1server_request id1_request;
    x_SetParams(seqref, id1_request.SetGetsefromgi(), is_snp);
    CObjectOStreamAsnBinary out(*stream);