1 #ifndef OSMIUM_IO_WRITER_HPP 2 #define OSMIUM_IO_WRITER_HPP 37 #include <osmium/io/detail/output_format.hpp> 38 #include <osmium/io/detail/queue_util.hpp> 39 #include <osmium/io/detail/read_write.hpp> 40 #include <osmium/io/detail/write_thread.hpp> 56 #include <initializer_list> 71 inline size_t get_output_queue_size() noexcept {
103 static constexpr
size_t default_buffer_size = 10 * 1024 * 1024;
107 detail::future_string_queue_type m_output_queue{detail::get_output_queue_size(),
"raw_output"};
109 std::unique_ptr<osmium::io::detail::OutputFormat> m_output{
nullptr};
113 size_t m_buffer_size = default_buffer_size;
115 std::future<bool> m_write_future{};
123 } m_status = status::okay;
126 static void write_thread(detail::future_string_queue_type& output_queue,
127 std::unique_ptr<osmium::io::Compressor>&& compressor,
128 std::promise<bool>&& write_promise) {
129 detail::WriteThread write_thread{output_queue,
130 std::move(compressor),
131 std::move(write_promise)};
136 if (buffer && buffer.committed() > 0) {
137 m_output->write_buffer(std::move(buffer));
143 if (m_buffer && m_buffer.committed() > 0) {
147 swap(m_buffer, buffer);
149 m_output->write_buffer(std::move(buffer));
153 template <
typename TFunction,
typename... TArgs>
155 if (m_status != status::okay) {
156 throw io_error(
"Can not write to writer when in status 'closed' or 'error'");
160 func(std::forward<TArgs>(args)...);
162 m_status = status::error;
163 detail::add_to_queue(m_output_queue, std::current_exception());
164 detail::add_end_of_data_to_queue(m_output_queue);
177 options.
pool = &pool;
189 options.
sync = value;
193 if (m_status == status::okay) {
194 ensure_cleanup([&](){
195 do_write(std::move(m_buffer));
196 m_output->write_end();
197 m_status = status::closed;
198 detail::add_end_of_data_to_queue(m_output_queue);
228 template <
typename... TArgs>
230 m_file(file.check()) {
234 (void)std::initializer_list<int>{
235 (set_option(options, args), 0)...
242 m_output = osmium::io::detail::OutputFormatFactory::instance().create_output(*options.pool, m_file, m_output_queue);
244 if (options.header.get(
"generator").empty()) {
248 std::unique_ptr<osmium::io::Compressor> compressor =
249 CompressionFactory::instance().create_compressor(file.
compression(),
250 osmium::io::detail::open_for_writing(m_file.
filename(), options.allow_overwrite),
253 std::promise<bool> write_promise;
254 m_write_future = write_promise.get_future();
257 ensure_cleanup([&](){
258 m_output->write_header(options.header);
262 template <
typename... TArgs>
263 explicit Writer(
const std::string& filename, TArgs&&... args) :
267 template <
typename... TArgs>
268 explicit Writer(
const char* filename, TArgs&&... args) :
290 return m_buffer_size;
298 m_buffer_size = size;
309 ensure_cleanup([&](){
323 ensure_cleanup([&](){
325 do_write(std::move(buffer));
337 ensure_cleanup([&](){
343 m_buffer.push_back(item);
346 m_buffer.push_back(item);
363 if (m_write_future.valid()) {
364 m_write_future.get();
374 #endif // OSMIUM_IO_WRITER_HPP fsync sync
Definition: writer.hpp:172
~Writer() noexcept
Definition: writer.hpp:278
std::size_t get_max_queue_size(const char *queue_name, std::size_t default_value) noexcept
Definition: config.hpp:83
Definition: writer.hpp:169
static Pool & default_instance()
Definition: pool.hpp:174
void do_write(osmium::memory::Buffer &&buffer)
Definition: writer.hpp:135
void do_flush()
Definition: writer.hpp:141
void swap(Buffer &lhs, Buffer &rhs)
Definition: buffer.hpp:755
void do_close()
Definition: writer.hpp:192
osmium::io::Header header
Definition: writer.hpp:170
void set_buffer_size(size_t size) noexcept
Definition: writer.hpp:297
static void write_thread(detail::future_string_queue_type &output_queue, std::unique_ptr< osmium::io::Compressor > &&compressor, std::promise< bool > &&write_promise)
Definition: writer.hpp:126
void operator()(osmium::memory::Buffer &&buffer)
Definition: writer.hpp:322
Namespace for everything in the Osmium library.
Definition: assembler.hpp:53
status
Definition: writer.hpp:119
#define LIBOSMIUM_VERSION_STRING
Definition: version.hpp:40
size_t buffer_size() const noexcept
Definition: writer.hpp:289
fsync
Definition: writer_options.hpp:51
static void set_option(options_type &options, fsync value)
Definition: writer.hpp:188
void close()
Definition: writer.hpp:360
void flush()
Definition: writer.hpp:308
osmium::io::File m_file
Definition: writer.hpp:105
static void set_option(options_type &options, osmium::thread::Pool &pool)
Definition: writer.hpp:176
Definition: buffer.hpp:97
Definition: buffer.hpp:58
const char * buffer() const noexcept
Definition: file.hpp:143
static void set_option(options_type &options, overwrite value)
Definition: writer.hpp:184
Definition: writer.hpp:101
Writer(const char *filename, TArgs &&... args)
Definition: writer.hpp:268
osmium::thread::Pool * pool
Definition: writer.hpp:173
Writer(const osmium::io::File &file, TArgs &&... args)
Definition: writer.hpp:229
void check_for_exception(std::future< T > &future)
Definition: util.hpp:55
Writer(const std::string &filename, TArgs &&... args)
Definition: writer.hpp:263
overwrite allow_overwrite
Definition: writer.hpp:171
static void set_option(options_type &options, const osmium::io::Header &header)
Definition: writer.hpp:180
file_compression compression() const noexcept
Definition: file.hpp:291
File & filename(const std::string &filename)
Definition: file.hpp:309
void operator()(const osmium::memory::Item &item)
Definition: writer.hpp:336
void ensure_cleanup(TFunction func, TArgs &&... args)
Definition: writer.hpp:154
overwrite
Definition: writer_options.hpp:43