BESDapResponseBuilder.cc

// -*- mode: c++; c-basic-offset:4 -*-

// This file is part of libdap, A C++ implementation of the OPeNDAP Data
// Access Protocol.

// Copyright (c) 2011 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 <signal.h>
#include <unistd.h>
#include <sys/stat.h>

#ifdef HAVE_UUID_UUID_H
#include <uuid/uuid.h>  // used to build CID header value for data ddx
#elif defined(HAVE_UUID_H)
#include <uuid.h>
#else
#error "Could not find UUID library header"
#endif


#ifndef WIN32
#include <sys/wait.h>
#else
#include <io.h>
#include <fcntl.h>
#include <process.h>
#endif

#include <iostream>
#include <string>
#include <sstream>
#include <fstream>

#include <cstring>
#include <ctime>

//#define DODS_DEBUG
#define CLEAR_LOCAL_DATA
#undef FUNCTION_CACHING
#undef USE_LOCAL_TIMEOUT_SCHEME

#include <DAS.h>
#include <DDS.h>
#include <Structure.h>
#include <ConstraintEvaluator.h>
#include <DDXParserSAX2.h>
#include <Ancillary.h>
#include <XDRStreamMarshaller.h>
#include <XDRFileUnMarshaller.h>

#include <DMR.h>
#include <D4Group.h>
#include <XMLWriter.h>
#include <D4AsyncUtil.h>
#include <D4StreamMarshaller.h>
#include <chunked_ostream.h>
#include <chunked_istream.h>
#include <D4ConstraintEvaluator.h>
#include <D4FunctionEvaluator.h>
#include <D4BaseTypeFactory.h>

#include <ServerFunctionsList.h>

#include <mime_util.h>  // for last_modified_time() and rfc_822_date()
#include <escaping.h>
#include <util.h>
#if USE_LOCAL_TIMEOUT_SCHEME
#ifndef WIN32
#include <SignalHandler.h>
#include <EventHandler.h>
#include <AlarmHandler.h>
#endif
#endif

#include "TheBESKeys.h"
#include "BESDapResponseBuilder.h"
#include "BESContextManager.h"
#include "BESDapResponseCache.h"
#include "BESStoredDapResultCache.h"
#include "BESUtil.h"
#include "BESDebug.h"
#include "BESStopWatch.h"
#include "DapFunctionUtils.h"

using namespace std;
using namespace libdap;

const string CRLF = "\r\n";             // Change here, expr-test.cc
const string BES_KEY_TIMEOUT_CANCEL = "BES.CancelTimeoutOnSend";

void BESDapResponseBuilder::initialize()
{
    bool found = false;
    string cancel_timeout_on_send = "";
    TheBESKeys::TheKeys()->get_value(BES_KEY_TIMEOUT_CANCEL, cancel_timeout_on_send, found);
    if (found && !cancel_timeout_on_send.empty()) {
        // The default value is false.
        downcase(cancel_timeout_on_send);
        if (cancel_timeout_on_send == "yes" || cancel_timeout_on_send == "true")
            d_cancel_timeout_on_send = true;
    }
}

BESDapResponseBuilder::~BESDapResponseBuilder()
{
#if USE_LOCAL_TIMEOUT_SCHEME
    // If an alarm was registered, delete it. The register code in SignalHandler
    // always deletes the old alarm handler object, so only the one returned by
    // remove_handler needs to be deleted at this point.
    delete dynamic_cast<AlarmHandler*>(SignalHandler::instance()->remove_handler(SIGALRM));
#endif
}

string BESDapResponseBuilder::get_ce() const
{
    return d_dap2ce;
}

void BESDapResponseBuilder::set_ce(string _ce)
{
    d_dap2ce = www2id(_ce, "%", "%20");
}

string BESDapResponseBuilder::get_dap4ce() const
{
    return d_dap4ce;
}

void BESDapResponseBuilder::set_dap4ce(string _ce)
{
    d_dap4ce = www2id(_ce, "%", "%20");
}

string BESDapResponseBuilder::get_dap4function() const
{
    return d_dap4function;
}

void BESDapResponseBuilder::set_dap4function(string _func)
{
    d_dap4function = www2id(_func, "%", "%20");
}

std::string BESDapResponseBuilder::get_store_result() const
{
    return d_store_result;
}

void BESDapResponseBuilder::set_store_result(std::string _sr)
{
    d_store_result = _sr;
    BESDEBUG("dap", "BESDapResponseBuilder::set_store_result() - store_result: " << _sr << endl);
}

std::string BESDapResponseBuilder::get_async_accepted() const
{
    return d_async_accepted;
}

void BESDapResponseBuilder::set_async_accepted(std::string _aa)
{
    d_async_accepted = _aa;
    BESDEBUG("dap", "BESDapResponseBuilder::set_async_accepted() - async_accepted: " << _aa << endl);
}

string BESDapResponseBuilder::get_dataset_name() const
{
    return d_dataset;
}

void BESDapResponseBuilder::set_dataset_name(const string ds)
{
    d_dataset = www2id(ds, "%", "%20");
}

void BESDapResponseBuilder::set_timeout(int t)
{
    d_timeout = t;
}

int BESDapResponseBuilder::get_timeout() const
{
    return d_timeout;
}

void
BESDapResponseBuilder::timeout_on() const
{
#if USE_LOCAL_TIMEOUT_SCHEME
#ifndef WIN32
    alarm(d_timeout);
#endif
#endif
}

void
BESDapResponseBuilder::timeout_off()
{
#if USE_LOCAL_TIMEOUT_SCHEME
#ifndef WIN32
    alarm(0);
#endif
#endif
}

void BESDapResponseBuilder::conditional_timeout_cancel()
{
    if (d_cancel_timeout_on_send)
        alarm(0);
}

void BESDapResponseBuilder::register_timeout() const
{
#if USE_LOCAL_TIMEOUT_SCHEME
#ifndef WIN32
    SignalHandler *sh = SignalHandler::instance();
    EventHandler *old_eh = sh->register_handler(SIGALRM, new AlarmHandler());
    delete old_eh;
#endif
#endif
}


void BESDapResponseBuilder::establish_timeout(ostream &) const
{
#if USE_LOCAL_TIMEOUT_SCHEME
#ifndef WIN32
    if (d_timeout > 0) {
        SignalHandler *sh = SignalHandler::instance();
        EventHandler *old_eh = sh->register_handler(SIGALRM, new AlarmHandler());
        delete old_eh;
        alarm(d_timeout);
    }
#endif
#endif
}

void BESDapResponseBuilder::remove_timeout() const
{
#if USE_LOCAL_TIMEOUT_SCHEME
    alarm(0);
#endif
}

static string::size_type find_closing_paren(const string &ce, string::size_type pos)
{
    // Iterate over the string finding all ( or ) characters until the matching ) is found.
    // For each ( found, increment count. When a ) is found and count is zero, it is the
    // matching closing paren, otherwise, decrement count and keep looking.
    int count = 1;
    do {
        pos = ce.find_first_of("()", pos + 1);
        if (pos == string::npos)
            throw Error(malformed_expr, "Expected to find a matching closing parenthesis in " + ce);

        if (ce[pos] == '(')
            ++count;
        else
            --count;    // must be ')'

    } while (count > 0);

    return pos;
}

void BESDapResponseBuilder::split_ce(ConstraintEvaluator &eval, const string &expr)
{
    BESDEBUG("dap", "BESDapResponseBuilder::split_ce() - source expression: " << expr << endl);

    string ce;
    if (!expr.empty())
        ce = expr;
    else
        ce = d_dap2ce;

    BESDEBUG("dap", "BESDapResponseBuilder::split_ce() - ce: " << ce << endl);

    string btp_function_ce = "";
    string::size_type pos = 0;

    // This hack assumes that the functions are listed first. Look for the first
    // open paren and the last closing paren to accommodate nested function calls
    string::size_type first_paren = ce.find("(", pos);
    string::size_type closing_paren = string::npos;
    if (first_paren != string::npos) closing_paren = find_closing_paren(ce, first_paren); //ce.find(")", pos);

    while (first_paren != string::npos && closing_paren != string::npos) {
        // Maybe a BTP function; get the name of the potential function
        string name = ce.substr(pos, first_paren - pos);

        // is this a BTP function
        btp_func f;
        if (eval.find_function(name, &f)) {
            // Found a BTP function
            if (!btp_function_ce.empty()) btp_function_ce += ",";
            btp_function_ce += ce.substr(pos, closing_paren + 1 - pos);
            ce.erase(pos, closing_paren + 1 - pos);
            if (ce[pos] == ',') ce.erase(pos, 1);
        }
        else {
            pos = closing_paren + 1;
            // exception?
            if (pos < ce.length() && ce.at(pos) == ',') ++pos;
        }

        first_paren = ce.find("(", pos);
        closing_paren = ce.find(")", pos);
    }

    d_dap2ce = ce;
    d_btp_func_ce = btp_function_ce;

    BESDEBUG("dap", "BESDapResponseBuilder::split_ce() - Modified constraint: " << d_dap2ce << endl);
    BESDEBUG("dap", "BESDapResponseBuilder::split_ce() - BTP Function part: " << btp_function_ce << endl);
    BESDEBUG("dap", "BESDapResponseBuilder::split_ce() - END" << endl);
}

void BESDapResponseBuilder::send_das(ostream &out, DAS &das, bool with_mime_headers) const
{
    if (with_mime_headers) set_mime_text(out, dods_das, x_plain, last_modified_time(d_dataset), "2.0");

    das.print(out);

    out << flush;
}

void BESDapResponseBuilder::send_das(ostream &out, DDS &dds, ConstraintEvaluator &eval, bool constrained,
    bool with_mime_headers)
{
#if USE_LOCAL_TIMEOUT_SCHEME
    // Set up the alarm.
    establish_timeout(out);
    dds.set_timeout(d_timeout);
#endif
    if (!constrained) {
        if (with_mime_headers) set_mime_text(out, dods_das, x_plain, last_modified_time(d_dataset), "2.0");

        conditional_timeout_cancel();

        dds.print_das(out);
        out << flush;

        return;
    }

    split_ce(eval);

    // If there are functions, parse them and eval.
    // Use that DDS and parse the non-function ce
    // Serialize using the second ce and the second dds
    if (!d_btp_func_ce.empty()) {
        DDS *fdds = 0;
        string cache_token = "";
        ConstraintEvaluator func_eval;
        BESDapResponseCache *responseCache = BESDapResponseCache::get_instance();

        if (responseCache) {
            fdds = responseCache->cache_dataset(dds, d_btp_func_ce, this, &func_eval, cache_token);
        }
        else {
            func_eval.parse_constraint(d_btp_func_ce, dds);
            fdds = func_eval.eval_function_clauses(dds);
        }

        if (with_mime_headers)
            set_mime_text(out, dods_das, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        fdds->print_das(out);

        if (responseCache)
                responseCache->unlock_and_close(cache_token);

        delete fdds;
    }
    else {
        eval.parse_constraint(d_dap2ce, dds); // Throws Error if the ce doesn't parse.

        if (with_mime_headers)
            set_mime_text(out, dods_das, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        dds.print_das(out);
    }

    out << flush;
}


#if 0

static bool ends_with (const string &full_string, const string &ending) {
    if (full_string.length() >= ending.length()) {
        return (0 == full_string.compare (full_string.length() - ending.length(), ending.length(), ending));
    } else {
        return false;
    }
}

static DDS *promote_function_output_structure(DDS *fdds)
{
    // Look in the top level of the DDS for a promotable member - i.e. a member
    // variable that is a collection and whose name ends with "_unwrap"
    bool found_promotable_member = false;
    for (DDS::Vars_citer di = fdds->var_begin(), de = fdds->var_end(); di != de && !found_promotable_member; ++di) {
        Structure *collection = dynamic_cast<Structure *>(*di);
        if (collection && ends_with(collection->name(), "_unwrap")) {
            found_promotable_member = true;
        }
    }

    // If we found one or more promotable member variables, promote them.
    if (found_promotable_member) {

        // Dump pointers to the values here temporarily... If we had methods in libdap
        // that could be used to access the underlying erase() and insert() methods, we
        // could skip the (maybe expensive) copy operations I use below. What we would
        // need are ways to delete a Structure/Constructor without calling delete on its
        // fields and ways to call vector::erase() and vector::insert(). Some of this
        // exists, but it's not quite enough.

        DDS *temp_dds = new DDS(fdds->get_factory(), fdds->get_dataset_name(), fdds->get_dap_version());

        for (DDS::Vars_citer di = fdds->var_begin(), de = fdds->var_end(); di != de; ++di) {
            Structure *collection = dynamic_cast<Structure *>(*di);
            if (collection && ends_with(collection->name(), "_unwrap")) {
                // So we're going to 'flatten this structure' and return its fields
                Structure::Vars_iter vi;
                for (vi =collection->var_begin(); vi != collection->var_end(); ++vi) {
                    temp_dds->add_var(*vi); // better to use add_var_nocopy(*vi); need to modify libdap?
                }
            }
            else {
                temp_dds->add_var(*di);
            }
        }

        delete fdds;
        return temp_dds;
    }
    else {
        // Otherwise do nothing to alter the DDS
        return fdds;
    }
}
#endif

void BESDapResponseBuilder::send_dds(ostream &out, DDS &dds, ConstraintEvaluator &eval, bool constrained,
    bool with_mime_headers)
{
    if (!constrained) {
        if (with_mime_headers)
            set_mime_text(out, dods_dds, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        dds.print(out);
        out << flush;
        return;
    }

#if USE_LOCAL_TIMEOUT_SCHEME
    // Set up the alarm.
    establish_timeout(out);
    dds.set_timeout(d_timeout);
#endif

    // Split constraint into two halves
    split_ce(eval);

    // If there are functions, parse them and eval.
    // Use that DDS and parse the non-function ce
    // Serialize using the second ce and the second dds
    if (!d_btp_func_ce.empty()) {
        string cache_token = "";
        DDS *fdds = 0;
        ConstraintEvaluator func_eval;

        BESDapResponseCache *responseCache = BESDapResponseCache::get_instance();

        if (responseCache) {
            fdds = responseCache->cache_dataset(dds, d_btp_func_ce, this, &func_eval, cache_token);
        }
        else {
            func_eval.parse_constraint(d_btp_func_ce, dds);
            fdds = func_eval.eval_function_clauses(dds);
        }

        // Server functions might mark variables to use their read()
        // methods. Clear that so the CE in d_dap2ce will control what is
        // sent. If that is empty (there was only a function call) all
        // of the variables in the intermediate DDS (i.e., the function
        // result) will be sent.
        fdds->mark_all(false);

        eval.parse_constraint(d_dap2ce, *fdds);

        if (with_mime_headers)
            set_mime_text(out, dods_dds, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        // This next step utilizes a well known static method (so really it's a function;),
        // promote_function_output_structures() to look for
        // one or more top level Structures whose name indicates (by way of ending with
        // "_uwrap") that their contents should be promoted (aka moved) to the top level.
        // This is in support of a hack around the current API where server side functions
        // may only return a single DAP object and not a collection of objects. The name suffix
        // "_unwrap" is used as a signal from the function to the the various response
        // builders and transmitters that the representation needs to be altered before
        // transmission, and that in fact is what happens in our friend
        // promote_function_output_structures()
        promote_function_output_structures(/*&*/fdds);

        conditional_timeout_cancel();


        fdds->print_constrained(out);

        if (responseCache)
                responseCache->unlock_and_close(cache_token);

        delete fdds;
    }
    else {
        eval.parse_constraint(d_dap2ce, dds); // Throws Error if the ce doesn't parse.

        if (with_mime_headers)
            set_mime_text(out, dods_dds, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        dds.print_constrained(out);
    }

    out << flush;
}

bool BESDapResponseBuilder::store_dap2_result(ostream &out, DDS &dds, ConstraintEvaluator &eval)
{
    if (get_store_result().empty()) return false;

    string serviceUrl = get_store_result();

    XMLWriter xmlWrtr;
    D4AsyncUtil d4au;

    // FIXME Keys should be read in initialize(). Also, I think the D4AsyncUtil should
    // be removed from libdap - it is much more about how the BES processes these kinds
    // of operations. Change this when working on the response caching for ODSIP. But...
    // do we really need to put the style sheet in the bes.conf file? Should it be baked
    // into the code (because we don't want people to change it)?
    bool found;
    string *stylesheet_ref = 0, ss_ref_value;
    TheBESKeys::TheKeys()->get_value(D4AsyncUtil::STYLESHEET_REFERENCE_KEY, ss_ref_value, found);
    if (found && ss_ref_value.length() > 0) {
        stylesheet_ref = &ss_ref_value;
    }

    BESStoredDapResultCache *resultCache = BESStoredDapResultCache::get_instance();
    if (resultCache == NULL) {

        string msg = "The Stored Result request cannot be serviced. ";
        msg += "Unable to acquire StoredResultCache instance. ";
        msg += "This is most likely because the StoredResultCache is not (correctly) configured.";

        BESDEBUG("dap", "[WARNING] " << msg << endl);

        d4au.writeD4AsyncResponseRejected(xmlWrtr, UNAVAILABLE, msg, stylesheet_ref);
        out << xmlWrtr.get_doc();
        out << flush;

        BESDEBUG("dap", "BESDapResponseBuilder::store_dap2_result() - Sent AsyncRequestRejected" << endl);
    }
    else if (get_async_accepted().length() != 0) {

        BESDEBUG("dap", "BESDapResponseBuilder::store_dap2_result() - serviceUrl="<< serviceUrl << endl);

        BESStoredDapResultCache *resultCache = BESStoredDapResultCache::get_instance();
        string storedResultId = "";
        storedResultId = resultCache->store_dap2_result(dds, get_ce(), this, &eval);

        BESDEBUG("dap",
            "BESDapResponseBuilder::store_dap2_result() - storedResultId='"<< storedResultId << "'" << endl);

        string targetURL = BESUtil::assemblePath(serviceUrl, storedResultId);
        BESDEBUG("dap", "BESDapResponseBuilder::store_dap2_result() - targetURL='"<< targetURL << "'" << endl);

        XMLWriter xmlWrtr;
        d4au.writeD4AsyncAccepted(xmlWrtr, 0, 0, targetURL, stylesheet_ref);
        out << xmlWrtr.get_doc();
        out << flush;

        BESDEBUG("dap", "BESDapResponseBuilder::store_dap2_result() - sent DAP4 AsyncAccepted response" << endl);
    }
    else {
        d4au.writeD4AsyncRequired(xmlWrtr, 0, 0, stylesheet_ref);
        out << xmlWrtr.get_doc();
        out << flush;

        BESDEBUG("dap", "BESDapResponseBuilder::store_dap2_result() - sent DAP4 AsyncRequired  response" << endl);
    }

    return true;

}

void BESDapResponseBuilder::serialize_dap2_data_dds(ostream &out, DDS &dds, ConstraintEvaluator &eval, bool ce_eval)
{
    BESStopWatch sw;
    if (BESISDEBUG(TIMING_LOG)) sw.start("BESDapResponseBuilder::serialize_dap2_data_dds", "");

    BESDEBUG("dap", "BESDapResponseBuilder::serialize_dap2_data_dds() - BEGIN" << endl);

    dds.print_constrained(out);
    out << "Data:\n";
    out << flush;

    XDRStreamMarshaller m(out);

    // This only has an effect when the timeout in BESInterface::execute_request()
    // is set. Otherwise it does nothing.
    conditional_timeout_cancel();


    // Send all variables in the current projection (send_p())
    for (DDS::Vars_iter i = dds.var_begin(); i != dds.var_end(); i++) {
        if ((*i)->send_p()) {
            (*i)->serialize(eval, dds, m, ce_eval);
#ifdef CLEAR_LOCAL_DATA
            (*i)->clear_local_data();
#endif
        }
    }

    BESDEBUG("dap", "BESDapResponseBuilder::serialize_dap2_data_dds() - END" << endl);
}

void BESDapResponseBuilder::serialize_dap2_data_ddx(ostream &out, DDS &dds, ConstraintEvaluator &eval,
    const string &boundary, const string &start, bool ce_eval)
{
    BESDEBUG("dap", __PRETTY_FUNCTION__ << " BEGIN" << endl);

    // Write the MPM headers for the DDX (text/xml) part of the response
    libdap::set_mime_ddx_boundary(out, boundary, start, dods_ddx, x_plain);

    // Make cid
    uuid_t uu;
    uuid_generate(uu);
    char uuid[37];
    uuid_unparse(uu, &uuid[0]);
    char domain[256];
    if (getdomainname(domain, 255) != 0 || strlen(domain) == 0) strncpy(domain, "opendap.org", 255);

    string cid = string(&uuid[0]) + "@" + string(&domain[0]);

    // Send constrained DDX with a data blob reference.
    // FIXME Comment CID passed but ignored jhrg 10/20/15
    dds.print_xml_writer(out, true, cid);

    // write the data part mime headers here
    set_mime_data_boundary(out, boundary, cid, dods_data_ddx /* old value dap4_data*/, x_plain);

    XDRStreamMarshaller m(out);

    conditional_timeout_cancel();


    // Send all variables in the current projection (send_p()).
    for (DDS::Vars_iter i = dds.var_begin(); i != dds.var_end(); i++) {
        if ((*i)->send_p()) {
            (*i)->serialize(eval, dds, m, ce_eval);
#ifdef CLEAR_LOCAL_DATA
            (*i)->clear_local_data();
#endif
        }
    }

    BESDEBUG("dap", __PRETTY_FUNCTION__ << " END" << endl);
}

void BESDapResponseBuilder::send_dap2_data(ostream &data_stream, DDS &dds, ConstraintEvaluator &eval,
    bool with_mime_headers)
{
    BESDEBUG("dap", "BESDapResponseBuilder::send_dap2_data() - BEGIN"<< endl);

#if USE_LOCAL_TIMEOUT_SCHEME
    // Set up the alarm.
    establish_timeout(data_stream);
    dds.set_timeout(d_timeout);
#endif

    // Split constraint into two halves
    split_ce(eval);

    // If there are functions, parse them and eval.
    // Use that DDS and parse the non-function ce
    // Serialize using the second ce and the second dds
    if (!get_btp_func_ce().empty()) {
        BESDEBUG("dap",
            "BESDapResponseBuilder::send_dap2_data() - Found function(s) in CE: " << get_btp_func_ce() << endl);

        string cache_token = "";
        DDS *fdds = 0;
        // Define a local ce evaluator so that the clause from the function parse
        // won't get treated like selection clauses later on when serialize is called
        // on the DDS (fdds)
        ConstraintEvaluator func_eval;
        BESDapResponseCache *responseCache = 0;

#if FUNCTION_CACHING
        responseCache = BESDapResponseCache::get_instance();
#endif

        if (responseCache) {
            BESDEBUG("dap",
                "BESDapResponseBuilder::send_dap2_data() - Using the cache for the server function CE" << endl);
            fdds = responseCache->cache_dataset(dds, get_btp_func_ce(), this, &func_eval, cache_token);
        }
        else {
            BESDEBUG("dap", "BESDapResponseBuilder::send_dap2_data() - Cache not found; (re)calculating" << endl);
            func_eval.parse_constraint(get_btp_func_ce(), dds);
            fdds = func_eval.eval_function_clauses(dds);
        }

        // Server functions might mark variables to use their read()
        // methods. Clear that so the CE in d_dap2ce will control what is
        // sent. If that is empty (there was only a function call) all
        // of the variables in the intermediate DDS (i.e., the function
        // result) will be sent.
        fdds->mark_all(false);

        // This next step utilizes a well known static method (so really it's a function;),
        // promote_function_output_structures() to look for
        // one or more top level Structures whose name indicates (by way of ending with
        // "_uwrap") that their contents should be promoted (aka moved) to the top level.
        // This is in support of a hack around the current API where server side functions
        // may only return a single DAP object and not a collection of objects. The name suffix
        // "_unwrap" is used as a signal from the function to the the various response
        // builders and transmitters that the representation needs to be altered before
        // transmission, and that in fact is what happens in our friend
        // promote_function_output_structures()
        promote_function_output_structures(fdds);

        eval.parse_constraint(get_ce(), *fdds);

        fdds->tag_nested_sequences(); // Tag Sequences as Parent or Leaf node.

        if (fdds->get_response_limit() != 0 && fdds->get_request_size(true) > fdds->get_response_limit()) {
            string msg = "The Request for " + long_to_string(dds.get_request_size(true) / 1024)
                + "KB is too large; requests for this user are limited to "
                + long_to_string(dds.get_response_limit() / 1024) + "KB.";
            throw Error(msg);
        }

        if (with_mime_headers)
            set_mime_binary(data_stream, dods_data, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

#if FUNCTION_CACHING
        // This means: if we are not supposed to store the result, then serialize it.
        if (!store_dap2_result(data_stream, dds, eval)) {
            serialize_dap2_data_dds(data_stream, *fdds, eval, true /* was 'false'. jhrg 3/10/15 */);
        }

        if (responseCache)
            responseCache->unlock_and_close(cache_token);
#else
        serialize_dap2_data_dds(data_stream, *fdds, eval, true /* was 'false'. jhrg 3/10/15 */);
#endif

        delete fdds;
    }
    else {
        BESDEBUG("dap", "BESDapResponseBuilder::send_dap2_data() - Simple constraint" << endl);

        eval.parse_constraint(get_ce(), dds); // Throws Error if the ce doesn't parse.

        dds.tag_nested_sequences(); // Tag Sequences as Parent or Leaf node.

        if (dds.get_response_limit() != 0 && dds.get_request_size(true) > dds.get_response_limit()) {
            string msg = "The Request for " + long_to_string(dds.get_request_size(true) / 1024)
                + "KB is too large; requests for this user are limited to "
                + long_to_string(dds.get_response_limit() / 1024) + "KB.";
            throw Error(msg);
        }

        if (with_mime_headers)
            set_mime_binary(data_stream, dods_data, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        // This means: if we are not supposed to store the result, then serialize it.
        if (!store_dap2_result(data_stream, dds, eval)) {
            serialize_dap2_data_dds(data_stream, dds, eval);
        }
    }

    data_stream << flush;

    BESDEBUG("dap", "BESDapResponseBuilder::send_dap2_data() - END"<< endl);

}


void BESDapResponseBuilder::send_ddx(ostream &out, DDS &dds, ConstraintEvaluator &eval, bool with_mime_headers)
{
    if (d_dap2ce.empty()) {
        if (with_mime_headers)
            set_mime_text(out, dods_ddx, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        dds.print_xml_writer(out, false /*constrained */, "");
        //dds.print(out);
        out << flush;
        return;
    }

#if USE_LOCAL_TIMEOUT_SCHEME
    // Set up the alarm.
    establish_timeout(out);
    dds.set_timeout(d_timeout);
#endif

    // Split constraint into two halves
    split_ce(eval);

    // If there are functions, parse them and eval.
    // Use that DDS and parse the non-function ce
    // Serialize using the second ce and the second dds
    if (!d_btp_func_ce.empty()) {
        string cache_token = "";
        DDS *fdds = 0;
        ConstraintEvaluator func_eval;
        BESDapResponseCache *responseCache = 0;

#if FUNCTION_CACHING
        responseCache = BESDapResponseCache::get_instance();
#endif

        if (responseCache) {
            fdds = responseCache->cache_dataset(dds, d_btp_func_ce, this, &func_eval, cache_token);
        }
        else {
            func_eval.parse_constraint(d_btp_func_ce, dds);
            fdds = func_eval.eval_function_clauses(dds);
        }

        // Server functions might mark variables to use their read()
        // methods. Clear that so the CE in d_dap2ce will control what is
        // sent. If that is empty (there was only a function call) all
        // of the variables in the intermediate DDS (i.e., the function
        // result) will be sent.
        fdds->mark_all(false);

        // This next step utilizes a well known static method (so really it's a function;),
        // promote_function_output_structures() to look for
        // one or more top level Structures whose name indicates (by way of ending with
        // "_uwrap") that their contents should be promoted (aka moved) to the top level.
        // This is in support of a hack around the current API where server side functions
        // may only return a single DAP object and not a collection of objects. The name suffix
        // "_unwrap" is used as a signal from the function to the the various response
        // builders and transmitters that the representation needs to be altered before
        // transmission, and that in fact is what happens in our friend
        // promote_function_output_structures()
        promote_function_output_structures(fdds);

        eval.parse_constraint(d_dap2ce, *fdds);

        if (with_mime_headers)
            set_mime_text(out, dods_ddx, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        fdds->print_xml_writer(out, true, "");

#if FUNCTION_CACHING
        if (responseCache)
            responseCache->unlock_and_close(cache_token);
#endif

        delete fdds;
    }
    else {
        eval.parse_constraint(d_dap2ce, dds); // Throws Error if the ce doesn't parse.

        if (with_mime_headers)
            set_mime_text(out, dods_ddx, x_plain, last_modified_time(d_dataset), dds.get_dap_version());

        conditional_timeout_cancel();


        // dds.print_constrained(out);
        dds.print_xml_writer(out, true, "");
    }

    out << flush;
}

void BESDapResponseBuilder::send_dmr(ostream &out, DMR &dmr, bool with_mime_headers)
{
    // If the CE is not empty, parse it. The projections, etc., are set as a side effect.
    // If the parser returns false, the expression did not parse. The parser may also
    // throw Error
    if (!d_dap4ce.empty()) {

        BESDEBUG("dap", "BESDapResponseBuilder::send_dmr() - Parsing DAP4 constraint: '"<< d_dap4ce << "'"<< endl);

        D4ConstraintEvaluator parser(&dmr);
        bool parse_ok = parser.parse(d_dap4ce);
        if (!parse_ok) throw Error(malformed_expr, "Constraint Expression (" + d_dap4ce + ") failed to parse.");
    }
    // with an empty CE, send everything. Even though print_dap4() and serialize()
    // don't need this, other code may depend on send_p being set. This may change
    // if DAP4 has a separate function evaluation phase. jhrg 11/25/13
    else {
        dmr.root()->set_send_p(true);
    }

    if (with_mime_headers) set_mime_text(out, dap4_dmr, x_plain, last_modified_time(d_dataset), dmr.dap_version());

    conditional_timeout_cancel();


    XMLWriter xml;
    dmr.print_dap4(xml, /*constrained &&*/!d_dap4ce.empty() /* true == constrained */);
    out << xml.get_doc() << flush;
}

void BESDapResponseBuilder::send_dap4_data_using_ce(ostream &out, DMR &dmr, bool with_mime_headers)
{
    if (!d_dap4ce.empty()) {
        D4ConstraintEvaluator parser(&dmr);
        bool parse_ok = parser.parse(d_dap4ce);
        if (!parse_ok) throw Error(malformed_expr, "Constraint Expression (" + d_dap4ce + ") failed to parse.");
    }
    // with an empty CE, send everything. Even though print_dap4() and serialize()
    // don't need this, other code may depend on send_p being set. This may change
    // if DAP4 has a separate function evaluation phase. jhrg 11/25/13
    else {
        dmr.root()->set_send_p(true);
    }

    if (dmr.response_limit() != 0 && dmr.request_size(true) > dmr.response_limit()) {
        string msg = "The Request for " + long_to_string(dmr.request_size(true) / 1024)
            + "MB is too large; requests for this user are limited to " + long_to_string(dmr.response_limit() / 1024)
            + "MB.";
        throw Error(msg);
    }

    if (!store_dap4_result(out, dmr)) {
        serialize_dap4_data(out, dmr, with_mime_headers);
    }
}

void BESDapResponseBuilder::send_dap4_data(ostream &out, DMR &dmr, bool with_mime_headers)
{
    // If a function was passed in with this request, evaluate it and use that DMR
    // for the remainder of this request.
    // TODO Add caching for these function invocations
    if (!d_dap4function.empty()) {
        D4BaseTypeFactory d4_factory;
        DMR function_result(&d4_factory, "function_results");

        // Function modules load their functions onto this list. The list is
        // part of libdap, not the BES.
        if (!ServerFunctionsList::TheList())
            throw Error(
                "The function expression could not be evaluated because there are no server functions defined on this server");

        D4FunctionEvaluator parser(&dmr, ServerFunctionsList::TheList());
        bool parse_ok = parser.parse(d_dap4function);
        if (!parse_ok) throw Error("Function Expression (" + d_dap4function + ") failed to parse.");

        parser.eval(&function_result);

        // Now use the results of running the functions for the remainder of the
        // send_data operation.
        send_dap4_data_using_ce(out, function_result, with_mime_headers);
    }
    else {
        send_dap4_data_using_ce(out, dmr, with_mime_headers);
    }
}

void BESDapResponseBuilder::serialize_dap4_data(std::ostream &out, libdap::DMR &dmr, bool with_mime_headers)
{
    BESDEBUG("dap", "BESDapResponseBuilder::serialize_dap4_data() - BEGIN" << endl);

    if (with_mime_headers) set_mime_binary(out, dap4_data, x_plain, last_modified_time(d_dataset), dmr.dap_version());

    // Write the DMR
    XMLWriter xml;
    dmr.print_dap4(xml, !d_dap4ce.empty());

    // now make the chunked output stream; set the size to be at least chunk_size
    // but make sure that the whole of the xml plus the CRLF can fit in the first
    // chunk. (+2 for the CRLF bytes).
    chunked_ostream cos(out, max((unsigned int) CHUNK_SIZE, xml.get_doc_size() + 2));

    conditional_timeout_cancel();

    // using flush means that the DMR and CRLF are in the first chunk.
    cos << xml.get_doc() << CRLF << flush;

    // Write the data, chunked with checksums
    D4StreamMarshaller m(cos);
    dmr.root()->serialize(m, dmr, !d_dap4ce.empty());
#ifdef CLEAR_LOCAL_DATA
    dmr.root()->clear_local_data();
#endif
    cos << flush;

    BESDEBUG("dap", "BESDapResponseBuilder::serialize_dap4_data() - END" << endl);
}

bool BESDapResponseBuilder::store_dap4_result(ostream &out, libdap::DMR &dmr)
{
    if (get_store_result().length() != 0) {
        string serviceUrl = get_store_result();

        D4AsyncUtil d4au;
        XMLWriter xmlWrtr;

        // FIXME See above comment for store dap2 result
        bool found;
        string *stylesheet_ref = 0, ss_ref_value;
        TheBESKeys::TheKeys()->get_value(D4AsyncUtil::STYLESHEET_REFERENCE_KEY, ss_ref_value, found);
        if (found && ss_ref_value.length() > 0) {
            stylesheet_ref = &ss_ref_value;
        }

        BESStoredDapResultCache *resultCache = BESStoredDapResultCache::get_instance();
        if (resultCache == NULL) {

            string msg = "The Stored Result request cannot be serviced. ";
            msg += "Unable to acquire StoredResultCache instance. ";
            msg += "This is most likely because the StoredResultCache is not (correctly) configured.";

            BESDEBUG("dap", "[WARNING] " << msg << endl);
            d4au.writeD4AsyncResponseRejected(xmlWrtr, UNAVAILABLE, msg, stylesheet_ref);
            out << xmlWrtr.get_doc();
            out << flush;
            BESDEBUG("dap", "BESDapResponseBuilder::store_dap4_result() - Sent AsyncRequestRejected" << endl);

            return true;
        }

        if (get_async_accepted().length() != 0) {

            BESDEBUG("dap", "BESDapResponseBuilder::store_dap4_result() - serviceUrl="<< serviceUrl << endl);

            string storedResultId = "";
            storedResultId = resultCache->store_dap4_result(dmr, get_ce(), this);

            BESDEBUG("dap",
                "BESDapResponseBuilder::store_dap4_result() - storedResultId='"<< storedResultId << "'" << endl);

            string targetURL = BESUtil::assemblePath(serviceUrl, storedResultId);
            BESDEBUG("dap", "BESDapResponseBuilder::store_dap4_result() - targetURL='"<< targetURL << "'" << endl);

            d4au.writeD4AsyncAccepted(xmlWrtr, 0, 0, targetURL, stylesheet_ref);
            out << xmlWrtr.get_doc();
            out << flush;
            BESDEBUG("dap", "BESDapResponseBuilder::store_dap4_result() - sent AsyncAccepted" << endl);

        }
        else {
            d4au.writeD4AsyncRequired(xmlWrtr, 0, 0, stylesheet_ref);
            out << xmlWrtr.get_doc();
            out << flush;
            BESDEBUG("dap", "BESDapResponseBuilder::store_dap4_result() - sent AsyncAccepted" << endl);
        }

        return true;
    }

    return false;
}