reader_pubseq.cpp

来自「ncbi源码」· C++ 代码 · 共 772 行 · 第 1/2 页

/*
 * ===========================================================================
 * PRODUCTION $Log: reader_pubseq.cpp,v $
 * PRODUCTION Revision 1000.1  2004/06/01 19:42:05  gouriano
 * PRODUCTION PRODUCTION: UPGRADED [GCC34_MSVC7] Dev-tree R1.52
 * PRODUCTION
 * ===========================================================================
 */

/*  $Id: reader_pubseq.cpp,v 1000.1 2004/06/01 19:42:05 gouriano Exp $
* ===========================================================================
*                            PUBLIC DOMAIN NOTICE
*               National Center for Biotechnology Information
*
*  This software/database is a "United States Government Work" under the
*  terms of the United States Copyright Act.  It was written as part of
*  the author's official duties as a United States Government employee and
*  thus cannot be copyrighted.  This software/database is freely available
*  to the public for use. The National Library of Medicine and the U.S.
*  Government have not placed any restriction on its use or reproduction.
*
*  Although all reasonable efforts have been taken to ensure the accuracy
*  and reliability of the software and data, the NLM and the U.S.
*  Government do not and cannot warrant the performance or results that
*  may be obtained by using this software or data. The NLM and the U.S.
*  Government disclaim all warranties, express or implied, including
*  warranties of performance, merchantability or fitness for any particular
*  purpose.
*
*  Please cite the author in any work or product based on this material.
* ===========================================================================
*
*  Author:  Anton Butanaev, Eugene Vasilchenko
*
*  File Description: Data reader from Pubseq_OS
*
*/

#include <ncbi_pch.hpp>
#include <objtools/data_loaders/genbank/readers/pubseqos/reader_pubseq.hpp>
#include <objtools/data_loaders/genbank/readers/pubseqos/seqref_pubseq.hpp>
#include <objtools/data_loaders/genbank/reader_snp.hpp>

#include <objmgr/objmgr_exception.hpp>
#include <objmgr/impl/tse_info.hpp>

#include <dbapi/driver/exception.hpp>
#include <dbapi/driver/driver_mgr.hpp>
#include <dbapi/driver/drivers.hpp>

#include <objects/seqloc/Seq_id.hpp>
#include <objects/seqset/Seq_entry.hpp>
#include <objects/seqset/Bioseq_set.hpp>
#include <objects/seq/Seq_annot.hpp>

#include <corelib/ncbicntr.hpp>
#include <corelib/plugin_manager_impl.hpp>

#include <serial/objostrasn.hpp>

#include <util/compress/reader_zlib.hpp>

#include <memory>

BEGIN_NCBI_SCOPE
BEGIN_SCOPE(objects)

#if !defined(HAVE_SYBASE_REENTRANT) && defined(NCBI_THREADS)
// we have non MT-safe library used in MT application
static CAtomicCounter s_pubseq_readers;
#endif

CPubseqReader::CPubseqReader(TConn noConn,
                             const string& server,
                             const string& user,
                             const string& pswd)
    : m_Server(server) , m_User(user), m_Password(pswd), m_Context(NULL),
      m_NoMoreConnections(false)
{
    noConn=1; // limit number of simultaneous connections to one
#if !defined(HAVE_SYBASE_REENTRANT)
    noConn=1;
#endif
#if defined(NCBI_THREADS) && !defined(HAVE_SYBASE_REENTRANT)
    if ( s_pubseq_readers.Add(1) > 1 ) {
        s_pubseq_readers.Add(-1);
        NCBI_THROW(CLoaderException, eNoConnection,
                   "Attempt to open multiple pubseq_readers "
                   "without MT-safe DB library");
    }
#endif
    try {
        SetParallelLevel(noConn);
    }
    catch ( ... ) {
        SetParallelLevel(0);
        throw;
    }
}


CPubseqReader::~CPubseqReader()
{
    SetParallelLevel(0);

#if !defined(HAVE_SYBASE_REENTRANT) && defined(NCBI_THREADS)
    s_pubseq_readers.Add(-1);
#endif
}


CReader::TConn CPubseqReader::GetParallelLevel(void) const
{
    return m_Pool.size();
}


void CPubseqReader::SetParallelLevel(TConn size)
{
    size_t oldSize = m_Pool.size();
    for(size_t i = size; i < oldSize; ++i) {
        delete m_Pool[i];
        m_Pool[i] = 0;
    }

    m_Pool.resize(size);

    for(size_t i = oldSize; i < min(1u, size); ++i) {
        m_Pool[i] = x_NewConnection();
    }
}


CDB_Connection* CPubseqReader::x_GetConnection(TConn conn)
{
    conn = conn % m_Pool.size();
    CDB_Connection* ret = m_Pool[conn];
    if ( !ret ) {
        ret = x_NewConnection();

        if ( !ret ) {
            NCBI_THROW(CLoaderException, eNoConnection,
                       "too many connections failed: probably server is dead");
        }

        m_Pool[conn] = ret;
    }
    return ret;
}


void CPubseqReader::Reconnect(TConn conn)
{
    LOG_POST("Reconnect");
    conn = conn % m_Pool.size();
    delete m_Pool[conn];
    m_Pool[conn] = 0;
}


CDB_Connection *CPubseqReader::x_NewConnection(void)
{
    for ( int i = 0; !m_NoMoreConnections && i < 3; ++i ) {
        if ( m_Context.get() == NULL ) {
            C_DriverMgr drvMgr;
            //DBAPI_RegisterDriver_CTLIB(drvMgr);
            //DBAPI_RegisterDriver_DBLIB(drvMgr);
            string errmsg;
            FDBAPI_CreateContext createContextFunc =
                drvMgr.GetDriver("ctlib",&errmsg);
            if ( !createContextFunc ) {
                LOG_POST(errmsg);
#if defined(HAVE_SYBASE_REENTRANT) && defined(NCBI_THREADS)
                m_NoMoreConnections = true;
                NCBI_THROW(CLoaderException, eNoConnection,
                           "Neither ctlib nor dblib are available");
#else
                createContextFunc = drvMgr.GetDriver("dblib",&errmsg);
                if ( !createContextFunc ) {
                    LOG_POST(errmsg);
                    m_NoMoreConnections = true;
                    NCBI_THROW(CLoaderException, eNoConnection,
                               "Neither ctlib nor dblib are available");
                }
#endif
            }
            map<string,string> args;
            args["packet"]="3584"; // 7*512
            m_Context.reset((*createContextFunc)(&args));
            //m_Context.reset((*createContextFunc)(0));
        }
        try {
            auto_ptr<CDB_Connection> conn(m_Context->Connect(m_Server,
                                                             m_User,
                                                             m_Password,
                                                             0));
            if ( conn.get() ) {
                auto_ptr<CDB_LangCmd> cmd(conn->LangCmd("set blob_stream on"));
                if ( cmd.get() ) {
                    cmd->Send();
                }
            }
            return conn.release();
        }
        catch ( CException& exc ) {
            ERR_POST("CPubseqReader::x_NewConnection: "
                     "cannot connect: " << exc.what());
        }
    }
    m_NoMoreConnections = true;
    return 0;
}


int CPubseqReader::ResolveSeq_id_to_gi(const CSeq_id& seqId, TConn conn)
{
    // note: this was
    //CDB_VarChar asnIn(static_cast<string>(CNcbiOstrstreamToString(oss)));
    // but MSVC doesn't like this.  This is the only version that
    // will compile:
    CDB_VarChar asnIn;
    {{
        CNcbiOstrstream oss;
        {{
            CObjectOStreamAsn ooss(oss);
            ooss << seqId;
        }}
        asnIn = CNcbiOstrstreamToString(oss);
    }}
    
    CDB_Connection* db_conn = x_GetConnection(conn);
    auto_ptr<CDB_RPCCmd> cmd(db_conn->RPC("id_gi_by_seqid_asn", 1));
    cmd->SetParam("@asnin", &asnIn);
    cmd->Send();

    while(cmd->HasMoreResults()) {
        auto_ptr<CDB_Result> result(cmd->Result());
        if (result.get() == 0  ||  result->ResultType() != eDB_RowResult)
            continue;
            
        while(result->Fetch()) {
            for(unsigned pos = 0; pos < result->NofItems(); ++pos) {
                const string& name = result->ItemName(pos);
                if (name == "gi") {
                    CDB_Int giFound;
                    result->GetItem(&giFound);
                    return giFound.Value();
                }
                else {
                    result->SkipItem();
                }
            }
        }
    }
    return 0;
}


void CPubseqReader::ResolveSeq_id(TSeqrefs& srs, const CSeq_id& id, TConn conn)
{
    _TRACE("ResolveSeq_id: " << id.AsFastaString());
    // note: this was
    //CDB_VarChar asnIn(static_cast<string>(CNcbiOstrstreamToString(oss)));
    // but MSVC doesn't like this.  This is the only version that
    // will compile:
    CDB_VarChar asnIn;
    {{
        CNcbiOstrstream oss;
        {{
            CObjectOStreamAsn ooss(oss);
            ooss << id;
        }}
        asnIn = CNcbiOstrstreamToString(oss);
    }}

    CDB_Connection* db_conn = x_GetConnection(conn);
    auto_ptr<CDB_RPCCmd> cmd(db_conn->RPC("id_gi_by_seqid_asn", 1));
    cmd->SetParam("@asnin", &asnIn);
    cmd->Send();

    x_RetrieveSeqrefs(srs, *cmd, 0);
}


void CPubseqReader::RetrieveSeqrefs(TSeqrefs& srs, int gi, TConn conn)
{
    _TRACE("RetrieveSeqrefs: " << gi);
    CDB_Int giIn(gi);

    CDB_Connection* db_conn = x_GetConnection(conn);
    auto_ptr<CDB_RPCCmd> cmd(db_conn->RPC("id_gi_class", 1));
    cmd->SetParam("@gi", &giIn);
    cmd->Send();

    x_RetrieveSeqrefs(srs, *cmd, gi);
}


void CPubseqReader::x_RetrieveSeqrefs(TSeqrefs& srs, CDB_RPCCmd& cmd, int gi)
{
    while(cmd.HasMoreResults()) {
        auto_ptr<CDB_Result> result(cmd.Result());
        if (result.get() == 0  ||  result->ResultType() != eDB_RowResult)
            continue;
            
        while(result->Fetch()) {
            CDB_Int giGot(gi);
            CDB_Int sat;
            CDB_Int satKey;
            CDB_Int extFeat;
            
            _TRACE("next fetch: " << result->NofItems() << " items");
            for ( unsigned pos = 0; pos < result->NofItems(); ++pos ) {
                const string& name = result->ItemName(pos);
                _TRACE("next item: " << name);
                if (name == "gi") {
                    result->GetItem(&giGot);
                    _TRACE("gi: "<<giGot.Value());
                }
                else if (name == "sat" ) {
                    result->GetItem(&sat);
                    _TRACE("sat: "<<sat.Value());
                }
                else if(name == "sat_key") {
                    result->GetItem(&satKey);
                    _TRACE("sat_key: "<<satKey.Value());
                }
                else if(name == "extra_feat" || name == "ext_feat") {
                    result->GetItem(&extFeat);
#ifdef _DEBUG
                    if ( extFeat.IsNULL() ) {
                        _TRACE("ext_feat = NULL");
                    }
                    else {
                        _TRACE("ext_feat = "<<extFeat.Value());
                    }
#endif
                }
                else {
                    result->SkipItem();
                }
            }

            _ASSERT(sat.Value() != eSatellite_SNP);
            srs.push_back(Ref(new CSeqref(giGot.Value(),
                                          sat.Value(), satKey.Value())));
            if ( TrySNPSplit() ) {
                if ( extFeat.IsNULL() ) {
                    AddSNPSeqref(srs, giGot.Value(), CSeqref::fPossible);
                }
                else if ( extFeat.Value() & 1 ) {
                    AddSNPSeqref(srs, giGot.Value());
                }
            }
        }
    }
}


CRef<CTSE_Info> CPubseqReader::GetTSEBlob(const CSeqref& seqref, TConn conn)
{
    CDB_Connection* db_conn = x_GetConnection(conn);
    auto_ptr<CDB_RPCCmd> cmd(x_SendRequest(seqref, db_conn, false));
    auto_ptr<CDB_Result> result(x_ReceiveData(*cmd));
    return x_ReceiveMainBlob(*result);
}


CRef<CSeq_annot_SNP_Info> CPubseqReader::GetSNPAnnot(const CSeqref& seqref,
                                                     TConn conn)
{
    CDB_Connection* db_conn = x_GetConnection(conn);
    auto_ptr<CDB_RPCCmd> cmd(x_SendRequest(seqref, db_conn, true));
    auto_ptr<CDB_Result> result(x_ReceiveData(*cmd));
    return x_ReceiveSNPAnnot(*result);
}


CDB_RPCCmd* CPubseqReader::x_SendRequest(const CSeqref& seqref,
                                         CDB_Connection* db_conn,
                                         bool is_snp)
{
    auto_ptr<CDB_RPCCmd> cmd(db_conn->RPC("id_get_asn", 4));
    CDB_Int giIn(seqref.GetGi());
    CDB_SmallInt satIn(seqref.GetSat());
    CDB_Int satKeyIn(seqref.GetSatKey());