stream_query_impl.hxx

Go to the documentation of this file.

/* Code for parts of pqxx::internal::stream_query.
 *
 * These definitions need to be in a separate file in order to iron out
 * circular dependencies between headers.
 */
#ifndef PQXX_INTERNAL_STREAM_QUERY_IMPL_HXX
#define PQXX_INTERNAL_STREAM_QUERY_IMPL_HXX
 
namespace pqxx::internal
{
template<typename... TYPE>
inline stream_query<TYPE...>::stream_query(
  transaction_base &tx, std::string_view query, conversion_context c) :
        transaction_focus{tx, "stream_query"},
        m_char_finder{get_finder(tx, c.loc)},
        m_ctx{c}
{
  auto const r{tx.exec(std::format("COPY ({}) TO STDOUT", query), m_ctx.loc)};
  r.expect_columns(sizeof...(TYPE), m_ctx.loc);
  r.expect_rows(0, m_ctx.loc);
  register_me();
}
 
 
template<typename... TYPE>
PQXX_RETURNS_NONNULL inline char_finder_func *
stream_query<TYPE...>::get_finder(transaction_base const &tx, sl loc)
{
  auto const group{tx.conn().get_encoding_group(loc)};
  return get_char_finder<'\t', '\\'>(group, loc);
}
 
 
// TODO: Replace with generator?  Could be faster (local vars vs. members).
 
template<typename... TYPE> class stream_query_iterator final
{
  using stream_t = stream_query<TYPE...>;
 
public:
  using value_type = std::tuple<TYPE...>;
  using difference_type = long;
 
  stream_query_iterator(stream_t &home, sl loc) :
          m_home(&home),
          m_line{typename stream_query<TYPE...>::line_handle(
            nullptr, pqxx::internal::pq::pqfreemem)},
          m_created_loc{loc}
  {
    consume_line(loc);
  }
 
  stream_query_iterator(stream_query_iterator const &) = delete;
  stream_query_iterator(stream_query_iterator &&) = delete;
  ~stream_query_iterator() = default;
 
  stream_query_iterator &operator=(stream_query_iterator const &) = delete;
  stream_query_iterator &operator=(stream_query_iterator &&) = delete;
 
 
  stream_query_iterator &operator++() &
  {
    assert(not done());
    consume_line(m_created_loc);
    return *this;
  }
 
  value_type operator*() const
  {
    return m_home->parse_line(std::string_view{m_line.get(), m_line_size});
  }
 
  bool operator==(stream_query_end_iterator) const noexcept { return done(); }
 
  bool operator!=(stream_query_end_iterator) const noexcept
  {
    return not done();
  }
 
  friend bool
  operator==(stream_query_end_iterator, stream_query_iterator const &i)
  {
    return i.done();
  }
 
  friend bool
  operator!=(stream_query_end_iterator, stream_query_iterator const &i)
  {
    return not i.done();
  }
 
private:
  [[nodiscard]] bool done() const noexcept { return m_home->done(); }
 
 
  void consume_line(sl loc) &
  {
    auto [line, size]{m_home->read_line(loc)};
    m_line = std::move(line);
    m_line_size = size;
    if (m_line)
    {
      // We know how many fields to expect.  Replace the newline at the end
      // with the field separator, so the parsing loop only needs to scan for a
      // tab, not a tab or a newline.
      char *const ptr{m_line.get()};
      assert(ptr[size] == '\n');
      ptr[size] = '\t';
    }
  }
 
  stream_t *const m_home = nullptr;
 
  typename stream_t::line_handle m_line;
 
  std::size_t m_line_size = 0u;
 
  sl const m_created_loc;
};
 
 
template<typename... TYPE> inline auto stream_query<TYPE...>::begin() &
{
  return stream_query_iterator<TYPE...>{*this, m_ctx.loc};
}
 
 
template<typename... TYPE>
inline std::pair<typename stream_query<TYPE...>::line_handle, std::size_t>
stream_query<TYPE...>::read_line(sl loc) &
{
  assert(not done());
 
  internal::gate::connection_stream_from gate{trans().conn()};
  try
  {
    auto line{gate.read_copy_line(loc)};
    if (not line.first) [[unlikely]]
    {
      // This is how we get told the iteration is finished.
      close();
    }
    return line;
  }
  catch (std::exception const &)
  {
    close();
    throw;
  }
}
} // namespace pqxx::internal
#endif