DmrppCommon.cc

// -*- mode: c++; c-basic-offset:4 -*-
 
// This file is part of the BES
 
// Copyright (c) 2016 OPeNDAP, Inc.
// Author: James Gallagher <jgallagher@opendap.org>
//
// 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 Street, Fifth Floor, Boston, MA  02110-1301  USA
//
// You can contact OPeNDAP, Inc. at PO Box 112, Saunderstown, RI. 02874-0112.
 
#include "config.h"
 
#include <string>
#include <sstream>
#include <vector>
#include <iterator>
#include <cstdlib>
#include <cstring>
 
#include <curl/curl.h>
 
#include <libdap/BaseType.h>
#include <libdap/D4Attributes.h>
#include <libdap/XMLWriter.h>
#include <libdap/util.h>
 
#define PUGIXML_NO_XPATH
#define PUGIXML_HEADER_ONLY
#include <pugixml.hpp>
 
#include "url_impl.h"
#include "BESIndent.h"
#include "BESDebug.h"
#include "BESUtil.h"
#include "BESInternalError.h"
 
#include "DmrppRequestHandler.h"
#include "DmrppCommon.h"
#include "Chunk.h"
#include "byteswap_compat.h"
 
using namespace std;
using namespace libdap;
 
#define prolog std::string("DmrppCommon::").append(__func__).append("() - ")
 
namespace dmrpp {
 
// Used with BESDEBUG
static const string dmrpp_3 = "dmrpp:3";
static const string dmrpp_4 = "dmrpp:4";
 
bool DmrppCommon::d_print_chunks = false;
string DmrppCommon::d_dmrpp_ns = "http://xml.opendap.org/dap/dmrpp/1.0.0#";
string DmrppCommon::d_ns_prefix = "dmrpp";
 
void join_threads(pthread_t threads[], unsigned int num_threads)
{
    int status;
    for (unsigned int i = 0; i < num_threads; ++i) {
        if (threads[i]) {
            BESDEBUG(dmrpp_3, "Join thread " << i << " after an exception was caught." << endl);
            string *error = NULL;
            if ((status = pthread_join(threads[i], (void **) &error)) < 0) {
                BESDEBUG(dmrpp_3, "Could not join thread " << i << ", " << strerror(status)<< endl);
            }
            else if (error != NULL) {
                BESDEBUG(dmrpp_3, "Joined thread " << i << ", error exit: " << *error << endl);
            }
            else {
                BESDEBUG(dmrpp_3, "Joined thread " << i << ", successful exit." << endl);
            }
        }
    }
}
 
void DmrppCommon::set_filter(const string &value) {
    if (DmrppRequestHandler::d_emulate_original_filter_order_behavior) {
        d_filters = "";
        if (value.find("shuffle") != string::npos)
            d_filters.append(" shuffle");
        if (value.find("deflate") != string::npos)
            d_filters.append(" deflate");
        if (value.find("fletcher32") != string::npos)
            d_filters.append(" fletcher32");
 
        BESUtil::removeLeadingAndTrailingBlanks(d_filters);
    }
    else {
        d_filters = value;
    }
}
 
void DmrppCommon::parse_chunk_dimension_sizes(const string &chunk_dims_string)
{
    d_chunk_dimension_sizes.clear();
 
    if (chunk_dims_string.empty()) return;
 
    string chunk_dims = chunk_dims_string;
    // If the input is anything other than integers and spaces, throw
    if (chunk_dims.find_first_not_of("1234567890 ") != string::npos)
        throw BESInternalError("while processing chunk dimension information, illegal character(s)", __FILE__, __LINE__);
 
    // TODO Rewrite this to use split. jhrg 5/2/22
    string space(" ");
    size_t strPos = 0;
    string strVal;
 
    // Are there spaces or multiple values?
    if (chunk_dims.find(space) != string::npos) {
        // Process space delimited content
        while ((strPos = chunk_dims.find(space)) != string::npos) {
            strVal = chunk_dims.substr(0, strPos);
            // TODO stoull (CDS uses uint64_t) jhrg 5/2/22
            d_chunk_dimension_sizes.push_back(strtol(strVal.c_str(), nullptr, 10));
            chunk_dims.erase(0, strPos + space.length());
        }
    }
 
    // If it's multivalued there's still one more value left to process
    // If it's single valued the same is true, so let's ingest that.
    d_chunk_dimension_sizes.push_back(strtol(chunk_dims.c_str(), nullptr, 10));
}
 
void DmrppCommon::ingest_compression_type(const string &compression_type_string)
{
    if (compression_type_string.empty()) return;
    set_filter(compression_type_string);
}
 
void DmrppCommon::ingest_byte_order(const string &byte_order_string) {
 
    if (byte_order_string.empty()) return;
 
    // Process content
    if (byte_order_string.compare("LE") == 0) {
        d_byte_order = "LE";
        d_twiddle_bytes = is_host_big_endian();
    } else {
        if (byte_order_string.compare("BE") == 0) {
            d_byte_order = "BE";
            d_twiddle_bytes = !(is_host_big_endian());
        } else {
            throw BESInternalError("Did not recognize byteOrder.", __FILE__, __LINE__);
        }
    }
}
 
 
unsigned long DmrppCommon::add_chunk(
        shared_ptr<http::url> data_url,
        const string &byte_order,
        unsigned long long size,
        unsigned long long offset,
        const string &position_in_array)
{
    vector<unsigned long long> cpia_vector;
    Chunk::parse_chunk_position_in_array_string(position_in_array, cpia_vector);
    return add_chunk(move(data_url), byte_order, size, offset, cpia_vector);
}
 
 
unsigned long DmrppCommon::add_chunk(
        shared_ptr<http::url> data_url,
        const string &byte_order,
        unsigned long long size,
        unsigned long long offset,
        const vector<unsigned long long> &position_in_array)
{
    std::shared_ptr<Chunk> chunk(new Chunk(move(data_url), byte_order, size, offset, position_in_array));
 
    d_chunks.push_back(chunk);
    return d_chunks.size();
}
 
unsigned long DmrppCommon::add_chunk(
        const string &byte_order,
        unsigned long long size,
        unsigned long long offset,
        const string &position_in_array)
{
    vector<unsigned long long> cpia_vector;
    Chunk::parse_chunk_position_in_array_string(position_in_array, cpia_vector);
    return add_chunk(byte_order, size, offset, cpia_vector);
}
 
unsigned long DmrppCommon::add_chunk(
        const string &byte_order,
        unsigned long long size,
        unsigned long long offset,
        const vector<unsigned long long> &position_in_array)
{
    shared_ptr<Chunk> chunk(new Chunk( byte_order, size, offset, position_in_array));
 
    d_chunks.push_back(chunk);
    return d_chunks.size();
}
 
unsigned long DmrppCommon::add_chunk(
        const string &byte_order,
        const string &fill_value,
        libdap::Type fv_type,
        unsigned long long chunk_size,
        const vector<unsigned long long> &position_in_array)
{
    shared_ptr<Chunk> chunk(new Chunk(byte_order, fill_value, fv_type, chunk_size, position_in_array));
 
    d_chunks.push_back(chunk);
    return d_chunks.size();
}
 
char *
DmrppCommon::read_atomic(const string &name)
{
    if (get_chunks_size() != 1)
        throw BESInternalError(string("Expected only a single chunk for variable ") + name, __FILE__, __LINE__);
 
    auto chunk = get_immutable_chunks()[0];
 
    chunk->read_chunk();
 
    return chunk->get_rbuf();
}
 
void
DmrppCommon::print_chunks_element(XMLWriter &xml, const string &name_space)
{
    // Start element "chunks" with dmrpp namespace and attributes:
    if (xmlTextWriterStartElementNS(xml.get_writer(), (const xmlChar*)name_space.c_str(), (const xmlChar*) "chunks", NULL) < 0)
        throw BESInternalError("Could not start chunks element.", __FILE__, __LINE__);
 
    if (!d_filters.empty())
        if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar*) "compressionType", (const xmlChar*) d_filters.c_str()) < 0)
            throw BESInternalError("Could not write compression attribute.", __FILE__, __LINE__);
 
    if (d_uses_fill_value && !d_fill_value_str.empty()) {
        if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar*) "fillValue", (const xmlChar*) d_fill_value_str.c_str()) < 0)
            throw BESInternalError("Could not write fillValue attribute.", __FILE__, __LINE__);
    }
 
    if(!d_chunks.empty()) {
        auto first_chunk = get_immutable_chunks().front();
        if (!first_chunk->get_byte_order().empty()) {
            if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar *) "byteOrder",
                                        (const xmlChar *) first_chunk->get_byte_order().c_str()) < 0)
            throw BESInternalError("Could not write attribute byteOrder", __FILE__, __LINE__);
        }
    }
 
    if (!d_chunk_dimension_sizes.empty()) { //d_chunk_dimension_sizes.size() > 0) {
        // Write element "chunkDimensionSizes" with dmrpp namespace:
        ostringstream oss;
        copy(d_chunk_dimension_sizes.begin(), d_chunk_dimension_sizes.end(), ostream_iterator<unsigned int>(oss, " "));
        string sizes = oss.str();
        sizes.erase(sizes.size() - 1, 1);    // trim the trailing space
 
        if (xmlTextWriterWriteElementNS(xml.get_writer(), (const xmlChar*) name_space.c_str(), (const xmlChar*) "chunkDimensionSizes", NULL,
            (const xmlChar*) sizes.c_str()) < 0) throw BESInternalError("Could not write chunkDimensionSizes attribute.", __FILE__, __LINE__);
    }
 
    // Start elements "chunk" with dmrpp namespace and attributes:
    for(auto chunk: get_immutable_chunks()) {
 
        if (xmlTextWriterStartElementNS(xml.get_writer(), (const xmlChar*)name_space.c_str(), (const xmlChar*) "chunk", NULL) < 0)
            throw BESInternalError("Could not start element chunk", __FILE__, __LINE__);
 
            // Get offset string:
            ostringstream offset;
            offset << chunk->get_offset();
            if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar *) "offset",
                                            (const xmlChar *) offset.str().c_str()) < 0)
                throw BESInternalError("Could not write attribute offset", __FILE__, __LINE__);
 
            // Get nBytes string:
            ostringstream nBytes;
            nBytes << chunk->get_size();
            if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar *) "nBytes",
                                            (const xmlChar *) nBytes.str().c_str()) < 0)
                throw BESInternalError("Could not write attribute nBytes", __FILE__, __LINE__);
 
        if (chunk->get_position_in_array().size() > 0) {
            // Get position in array string:
            vector<unsigned long long> pia = chunk->get_position_in_array();
            ostringstream oss;
            oss << "[";
            copy(pia.begin(), pia.end(), ostream_iterator<unsigned int>(oss, ","));
            string pia_str = oss.str();
            if (pia.size() > 0) pia_str.replace(pia_str.size() - 1, 1, "]"); // replace the trailing ',' with ']'
            if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar*) "chunkPositionInArray", (const xmlChar*) pia_str.c_str()) < 0)
                throw BESInternalError("Could not write attribute position in array", __FILE__, __LINE__);
        }
 
        // End element "chunk":
        if (xmlTextWriterEndElement(xml.get_writer()) < 0) throw BESInternalError("Could not end chunk element", __FILE__, __LINE__);
    }
 
    if (xmlTextWriterEndElement(xml.get_writer()) < 0) throw BESInternalError("Could not end chunks element", __FILE__, __LINE__);
}
 
void
DmrppCommon::print_compact_element(XMLWriter &xml, const string &name_space, const std::string &encoded)
{
    // Write element "compact" with dmrpp namespace:
    ostringstream oss;
    copy(encoded.begin(), encoded.end(), ostream_iterator<char>(oss, ""));
    string sizes = oss.str();
 
    if (xmlTextWriterWriteElementNS(xml.get_writer(), (const xmlChar *) name_space.c_str(),
                                    (const xmlChar *) "compact", NULL,
                                    (const xmlChar *) sizes.c_str()) < 0)
        throw BESInternalError("Could not write compact element.", __FILE__, __LINE__);
}
 
void DmrppCommon::print_dmrpp(XMLWriter &xml, bool constrained /*false*/)
{
    BaseType &bt = dynamic_cast<BaseType&>(*this);
    if (constrained && !bt.send_p())
        return;
 
    if (xmlTextWriterStartElement(xml.get_writer(), (const xmlChar*)bt.type_name().c_str()) < 0)
        throw InternalErr(__FILE__, __LINE__, "Could not write " + bt.type_name() + " element");
 
    if (!bt.name().empty())
        if (xmlTextWriterWriteAttribute(xml.get_writer(), (const xmlChar*) "name", (const xmlChar*)bt.name().c_str()) < 0)
            throw InternalErr(__FILE__, __LINE__, "Could not write attribute for name");
 
    if (bt.is_dap4())
        bt.attributes()->print_dap4(xml);
 
    if (!bt.is_dap4() && bt.get_attr_table().get_size() > 0)
        bt.get_attr_table().print_xml_writer(xml);
 
    // This is the code added to libdap::BaseType::print_dap4(). jhrg 5/10/18
    if (DmrppCommon::d_print_chunks && get_chunks_size() > 0)
        print_chunks_element(xml, DmrppCommon::d_ns_prefix);
 
    if (xmlTextWriterEndElement(xml.get_writer()) < 0)
        throw InternalErr(__FILE__, __LINE__, "Could not end " + bt.type_name() + " element");
}
 
void DmrppCommon::dump(ostream & strm) const
{
    strm << BESIndent::LMarg << "is_filters_empty:             " << (is_filters_empty() ? "true" : "false") << endl;
    strm << BESIndent::LMarg << "filters: " << (d_filters.c_str()) << endl;
 
    const vector<unsigned long long> &chunk_dim_sizes = get_chunk_dimension_sizes();
 
    strm << BESIndent::LMarg << "chunk dimension sizes:  [";
    for (unsigned int i = 0; i < chunk_dim_sizes.size(); i++) {
        strm << (i ? "][" : "") << chunk_dim_sizes[i];
    }
    strm << "]" << endl;
 
    auto chunk_refs = get_immutable_chunks();
    strm << BESIndent::LMarg << "Chunks (aka chunks):" << (chunk_refs.size() ? "" : "None Found.") << endl;
    BESIndent::Indent();
    for (auto & chunk_ref : chunk_refs) {
        strm << BESIndent::LMarg;
        chunk_ref->dump(strm);
        strm << endl;
    }
}
 
void
DmrppCommon::load_chunks(BaseType *btp) {
    d_dmz->load_chunks(btp);
}
 
void
DmrppCommon::load_attributes(libdap::BaseType *btp)
{
    d_dmz->load_attributes(btp);
}
 
} // namespace dmrpp