mapnik/plugins/input/tiles/xyz_featureset.cpp

/*****************************************************************************
 *
 * This file is part of Mapnik (c++ mapping toolkit)
 *
 * Copyright (C) 2025 Artem Pavlenko
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 *****************************************************************************/

// mapnik
#include <mapnik/well_known_srs.hpp>
#include "xyz_featureset.hpp"
#include "vector_tile_compression.hpp"
// boost
#include <boost/format.hpp>

xyz_featureset::xyz_featureset(std::string url_template,
                               mapnik::context_ptr const& ctx,
                               int const zoom,
                               int const xmin,
                               int const xmax,
                               int const ymin,
                               int const ymax,
                               //mapnik::box2d<double> const& extent,
                               std::string const& layer,
                               std::unordered_map<std::string, std::string>& vector_tile_cache,
                               std::size_t datasource_hash)
    : url_template_(url_template),
      context_(ctx),
      zoom_(zoom),
      xmin_(xmin),
      xmax_(xmax),
      ymin_(ymin),
      ymax_(ymax),
      layer_(layer),
      vector_tile_(nullptr),
      vector_tile_cache_(vector_tile_cache),
      QUEUE_SIZE_((xmax - xmin + 1)*(ymax - ymin + 1)),
      queue_(QUEUE_SIZE_),
      stash_(ioc_,targets_, queue_),
      datasource_hash_(datasource_hash)
{
    boost::urls::url url = boost::urls::format(url_template_, {{"z", zoom_}, {"x", 0}, {"y", 0}});
    host_ = url.host();
    auto scheme = url.scheme();
    if (scheme == "https")
    {
        ssl_ = true;
        port_ = url.port().empty() ? "443" : url.port();
    }
    else
    {
        port_ = url.port().empty() ? "80" : url.port();
    }

}

xyz_featureset::~xyz_featureset()
{
    for(std::size_t i = 0; i < workers_.size(); ++i)
    {
        workers_[i].join();
    }

}

bool xyz_featureset::valid() const
{
    return vector_tile_.get() != nullptr;
}

mapnik::feature_ptr xyz_featureset::next_feature()
{
    mapnik::feature_ptr f = mapnik::feature_ptr();
    if (valid())
    {
        f = vector_tile_->next();
    }
    return f;
}

mapnik::feature_ptr xyz_featureset::next()
{
    mapnik::feature_ptr f = next_feature();
    if (f)
    {
        return f;
    }
    while (next_tile() && valid())
    {
        f = next_feature();
        if (f)
        {
            return f;
        }
    }
    return mapnik::feature_ptr();
}

void little_nap(std::chrono::microseconds us)
{
    auto start = std::chrono::high_resolution_clock::now();
    auto end = start + us;
    do
    {
        std::this_thread::yield();
    }
    while (std::chrono::high_resolution_clock::now() < end);
}

bool xyz_featureset::next_tile()
{
    if (first_)
    {
        first_ = false;
        for (int x = xmin_; x <= xmax_; ++x)
        {
            for (int y = ymin_; y <= ymax_; ++y)
            {
                ++num_tiles_;
                auto datasource_key = (boost::format("%1%-%2%-%3%-%4%") % datasource_hash_ % zoom_ % x % y).str();
                auto itr = vector_tile_cache_.find(datasource_key);
                if (itr == vector_tile_cache_.end())
                {
                    stash_.targets().emplace_back(zoom_, x, y);
                }
                else
                {
                    std::string buffer = itr->second;
                    stash_.push_async(tile_data(zoom_, x, y, std::move(buffer)));
                }
            }
        }
        if (num_tiles_ != QUEUE_SIZE_) throw mapnik::datasource_exception("FAIL");

        std::size_t max_threads = 4;
        std::size_t threads = std::min(max_threads, stash_.targets().size()) ;
        workers_.reserve(threads + 1);
        for(std::size_t i = 0; i < threads; ++i)
        {
            auto reporting_work = boost::asio::require(
                ioc_.get_executor(),
                boost::asio::execution::outstanding_work.tracked);
            if (ssl_)
            {
                workers_.emplace_back([this, reporting_work] {
                    boost::asio::io_context ioc;
                    std::make_shared<worker_ssl>(ioc, ssl_ctx_, url_template_, stash_, std::ref(done_))->run(host_, port_);
                    ioc.run();
                });
            }
            else
            {
                workers_.emplace_back([this, reporting_work] {
                    boost::asio::io_context ioc;
                    std::make_shared<worker>(ioc, url_template_, stash_, std::ref(done_))->run(host_, port_);
                    ioc.run();
                });
            }
        }
        workers_.emplace_back([this]
        {
            ioc_.run();
        });
    }
    // consume tiles from the queue
    bool status = false;
    //std::thread consumer([this, &status] {
        //std::cerr << "\e[1;41m Consumer thread:" << std::this_thread::get_id() << " layer:" << layer_ << "\e[0m" << std::endl;
        //std::size_t count = 0;
        while (!done_.load())
        {
            tile_data tile;
            if (queue_.pop(tile))
            {
                ++consumed_count_;
                if (tile.data.size() == 0)
                    break;
                status = true;

                //std::cerr << "\e[1;41m Consumer thread:" << std::this_thread::get_id() << " tile size:" << tile.data.size() << "\e[0m" << std::endl;


                auto datasource_key = (boost::format("%1%-%2%-%3%-%4%") % datasource_hash_ % tile.zoom % tile.x % tile.y).str();
                auto itr = vector_tile_cache_.find(datasource_key);
                if (itr == vector_tile_cache_.end())
                {
                    std::string decompressed;
                    mapnik::vector_tile_impl::zlib_decompress(tile.data.data(), tile.data.size(), decompressed);
                    vector_tile_cache_.emplace(datasource_key, decompressed);
                    vector_tile_.reset(new mvt_io(std::move(decompressed), context_, tile.x, tile.y, zoom_, layer_));
                }
                else
                {
                    std::string buffer = itr->second;
                    vector_tile_.reset(new mvt_io(std::move(buffer), context_, tile.x, tile.y, zoom_, layer_));
                }
                if (consumed_count_ == QUEUE_SIZE_) done_.store(true);
                break;
            }
            if (consumed_count_ == num_tiles_) done_.store(true);
            //++count;
            //little_nap(std::chrono::microseconds(100));
            //std::cerr << "\e[1;41m count:" <<  count << "\e[0m" << std::endl;
        }
        //status = false;
        //});

        //consumer.join();
    return status;
}