pipeline.hxx

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 *   FILE
 *      pqxx/pipeline.hxx
 *
 *   DESCRIPTION
 *      definition of the pqxx::pipeline class.
 *   Throughput-optimized query manager
 *   DO NOT INCLUDE THIS FILE DIRECTLY; include pqxx/pipeline instead.
 *
 * Copyright (c) 2003-2004, Jeroen T. Vermeulen <jtv@xs4all.nl>
 *
 * See COPYING for copyright license.  If you did not receive a file called
 * COPYING with this source code, please notify the distributor of this mistake,
 * or contact the author.
 *
 *-------------------------------------------------------------------------
 */
#include "pqxx/libcompiler.h"

#include <map>
#include <string>

#include "pqxx/transaction_base"


/* Methods tested in eg. self-test program test001 are marked with "//[t1]"
 */

namespace pqxx
{


class PQXX_LIBEXPORT pipeline : public internal::transactionfocus
{
public:
  typedef long query_id;

  explicit pipeline(transaction_base &, const PGSTD::string &PName=""); //[t69]

  ~pipeline() throw ();


  query_id insert(const PGSTD::string &);                               //[t69]

  void complete();                                                      //[t71]


  void flush();                                                         //[t70]

  bool is_finished(query_id) const;                                     //[t71]


  result retrieve(query_id qid)                                         //[t71]
        { return retrieve(m_queries.find(qid)).second; }


  PGSTD::pair<query_id, result> retrieve();                             //[t69]

  bool empty() const throw () { return m_queries.empty(); }             //[t69]


  int retain(int retain_max=2);                                         //[t70]


  void resume();                                                        //[t70]

private:
  void invariant() const;

  class Query
  {
  public:
    explicit Query(const PGSTD::string &q) : m_query(q), m_res() {}

    const result &get_result() const throw () { return m_res; }
    void set_result(const result &r) throw () { m_res = r; }
    const PGSTD::string &get_query() const throw () { return m_query; }

  private:
    PGSTD::string m_query;
    result m_res;
  };

  typedef PGSTD::map<query_id,Query> QueryMap;

  static query_id qid_limit() throw ()
  {
#if defined(PQXX_HAVE_LIMITS) && !defined(_MSC_VER)
    return PGSTD::numeric_limits<query_id>::max();
#else
    return LONG_MAX;
#endif
  }

  query_id generate_id();
    
  bool have_pending() const throw () 
        { return m_issuedrange.second != m_issuedrange.first; }

  void issue();

  void set_error_at(query_id qid) throw () { if (qid < m_error) m_error = qid; }

  void internal_error(const PGSTD::string &err) throw (PGSTD::logic_error);

  bool obtain_result(bool expect_none=false);

  void obtain_dummy();
  void get_further_available_results();
  void check_end_results();

  void receive_if_available();

  void receive(QueryMap::const_iterator stop);
  PGSTD::pair<query_id, result> retrieve(QueryMap::iterator);

  QueryMap m_queries;
  PGSTD::pair<QueryMap::iterator,QueryMap::iterator> m_issuedrange;
  int m_retain;
  int m_num_waiting;
  query_id m_q_id;

  bool m_dummy_pending;

  query_id m_error;

  pipeline(const pipeline &);
  pipeline &operator=(const pipeline &);
};


} // namespace

---