http_msg.h

/***
* ==++==
*
* Copyright (c) Microsoft Corporation. All rights reserved.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* HTTP Library: Request and reply message definitions.
*
* For the latest on this and related APIs, please see http://casablanca.codeplex.com.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#include <map>
#include <memory>
#include <string>
#include <vector>
#include <system_error>

#include "pplx/pplxtasks.h"
#include "cpprest/json.h"
#include "cpprest/uri.h"
#include "cpprest/http_headers.h"
#include "cpprest/details/cpprest_compat.h"
#include "cpprest/asyncrt_utils.h"
#include "cpprest/streams.h"
#include "cpprest/containerstream.h"

namespace web
{
namespace http
{

// URI class has been moved from web::http namespace to web namespace.
// The below using declarations ensure we don't break existing code.
// Please use the web::uri class going forward.
using web::uri;
using web::uri_builder;

namespace client
{
    class http_client;
}

typedef utility::string_t method;

class methods
{
public:
#define _METHODS
#define DAT(a,b) _ASYNCRTIMP const static method a;
#include "cpprest/details/http_constants.dat"
#undef _METHODS
#undef DAT
};

typedef unsigned short status_code;

class status_codes
{
public:
#define _PHRASES
#define DAT(a,b,c) const static status_code a=b;
#include "cpprest/details/http_constants.dat"
#undef _PHRASES
#undef DAT
};

namespace message_direction
{
    enum direction {
        upload,
        download
    };
}

typedef utility::string_t reason_phrase;
typedef std::function<void(message_direction::direction, utility::size64_t)> progress_handler;

struct http_status_to_phrase
{
    unsigned short id;
    reason_phrase phrase;
};

class header_names
{
public:
#define _HEADER_NAMES
#define DAT(a,b) _ASYNCRTIMP const static utility::string_t a;
#include "cpprest/details/http_constants.dat"
#undef _HEADER_NAMES
#undef DAT
};

class http_exception : public std::exception
{
public:

    http_exception(const utility::string_t &whatArg)
        : m_msg(utility::conversions::to_utf8string(whatArg)) {}

#ifdef _WIN32
    http_exception(std::string whatArg) : m_msg(std::move(whatArg)) {}
#endif

    http_exception(int errorCode)
        : m_errorCode(utility::details::create_error_code(errorCode))
    {
        m_msg = m_errorCode.message();
    }

    http_exception(int errorCode, const utility::string_t &whatArg)
        : m_errorCode(utility::details::create_error_code(errorCode)),
          m_msg(utility::conversions::to_utf8string(whatArg))
    {}

#ifdef _WIN32
    http_exception(int errorCode, std::string whatArg) :
        m_errorCode(utility::details::create_error_code(errorCode)),
        m_msg(std::move(whatArg))
    {}
#endif

    http_exception(int errorCode, const std::error_category &cat) : m_errorCode(std::error_code(errorCode, cat))
    {
        m_msg = m_errorCode.message();
    }

    const char* what() const CPPREST_NOEXCEPT
    {
        return m_msg.c_str();
    }

    const std::error_code & error_code() const
    {
        return m_errorCode;
    }

private:
    std::error_code m_errorCode;
    std::string m_msg;
};

namespace details
{

class http_msg_base
{
public:

    friend class http::client::http_client;

    _ASYNCRTIMP http_msg_base();

    virtual ~http_msg_base() {}

    http_headers &headers() { return m_headers; }

    _ASYNCRTIMP void set_body(const concurrency::streams::istream &instream, const utf8string &contentType);
    _ASYNCRTIMP void set_body(const concurrency::streams::istream &instream, const utf16string &contentType);
    _ASYNCRTIMP void set_body(const concurrency::streams::istream &instream, utility::size64_t contentLength, const utf8string &contentType);
    _ASYNCRTIMP void set_body(const concurrency::streams::istream &instream, utility::size64_t contentLength, const utf16string &contentType);

    utility::string_t parse_and_check_content_type(bool ignore_content_type, const std::function<bool(const utility::string_t&)> &check_content_type);

    _ASYNCRTIMP utf8string extract_utf8string(bool ignore_content_type = false);
    _ASYNCRTIMP utf16string extract_utf16string(bool ignore_content_type = false);
    _ASYNCRTIMP utility::string_t extract_string(bool ignore_content_type = false);

    _ASYNCRTIMP json::value _extract_json(bool ignore_content_type = false);
    _ASYNCRTIMP std::vector<unsigned char> _extract_vector();

    virtual _ASYNCRTIMP utility::string_t to_string() const;

    virtual _ASYNCRTIMP void _complete(utility::size64_t bodySize, const std::exception_ptr &exceptionPtr = std::exception_ptr());

    void set_instream(const concurrency::streams::istream &instream)  { m_inStream = instream; }

    const concurrency::streams::istream & instream() const { return m_inStream; }

    void set_outstream(const concurrency::streams::ostream &outstream, bool is_default)  { m_outStream = outstream; m_default_outstream = is_default; }

    const concurrency::streams::ostream & outstream() const { return m_outStream; }

    const pplx::task_completion_event<utility::size64_t> & _get_data_available() const { return m_data_available; }

    _ASYNCRTIMP void _prepare_to_receive_data();

    _ASYNCRTIMP size_t _get_content_length();

protected:

    concurrency::streams::istream m_inStream;

    concurrency::streams::ostream m_outStream;

    http_headers m_headers;
    bool m_default_outstream;

    pplx::task_completion_event<utility::size64_t> m_data_available;
};

class _http_server_context
{
public:
    _http_server_context() {}
    virtual ~_http_server_context() {}
private:
};

class _http_response : public http::details::http_msg_base
{
public:
    _http_response() : m_status_code((std::numeric_limits<uint16_t>::max)()) { }

    _http_response(http::status_code code) : m_status_code(code) {}

    http::status_code status_code() const { return m_status_code; }

    void set_status_code(http::status_code code) { m_status_code = code; }

    const http::reason_phrase & reason_phrase() const { return m_reason_phrase; }

    void set_reason_phrase(const http::reason_phrase &reason) { m_reason_phrase = reason; }

    _ASYNCRTIMP utility::string_t to_string() const;

    _http_server_context * _get_server_context() const { return m_server_context.get(); }

    void _set_server_context(std::unique_ptr<details::_http_server_context> server_context) { m_server_context = std::move(server_context); }

private:
    std::unique_ptr<_http_server_context> m_server_context;

    http::status_code m_status_code;
    http::reason_phrase m_reason_phrase;
};

} // namespace details


class http_response
{
public:

    http_response() : _m_impl(std::make_shared<details::_http_response>()) { }

    http_response(http::status_code code)
        : _m_impl(std::make_shared<details::_http_response>(code)) { }

    http::status_code status_code() const { return _m_impl->status_code(); }

    void set_status_code(http::status_code code) const { _m_impl->set_status_code(code); }

    const http::reason_phrase & reason_phrase() const { return _m_impl->reason_phrase(); }

    void set_reason_phrase(const http::reason_phrase &reason) const { _m_impl->set_reason_phrase(reason); }

    http_headers &headers() { return _m_impl->headers(); }

    const http_headers &headers() const { return _m_impl->headers(); }

    utility::string_t to_string() const { return _m_impl->to_string(); }

    pplx::task<utility::string_t> extract_string(bool ignore_content_type = false) const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_string(ignore_content_type); });
    }

    pplx::task<utf8string> extract_utf8string(bool ignore_content_type = false) const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_utf8string(ignore_content_type); });
    }

    pplx::task<utf16string> extract_utf16string(bool ignore_content_type = false) const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_utf16string(ignore_content_type); });
    }

    pplx::task<json::value> extract_json(bool ignore_content_type = false) const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->_extract_json(ignore_content_type); });
    }

    pplx::task<std::vector<unsigned char>> extract_vector() const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl](utility::size64_t) { return impl->_extract_vector(); });
    }

    void set_body(utf8string &&body_text, const utf8string &content_type = utf8string("text/plain; charset=utf-8"))
    {
        const auto length = body_text.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream<std::string>(std::move(body_text)), length, content_type);
    }

    void set_body(const utf8string &body_text, const utf8string &content_type = utf8string("text/plain; charset=utf-8"))
    {
        _m_impl->set_body(concurrency::streams::bytestream::open_istream<std::string>(body_text), body_text.size(), content_type);
    }

    void set_body(const utf16string &body_text, utf16string content_type = ::utility::conversions::to_utf16string("text/plain"))
    {
        if (content_type.find(::utility::conversions::to_utf16string("charset=")) != content_type.npos)
        {
            throw std::invalid_argument("content_type can't contain a 'charset'.");
        }

        auto utf8body = utility::conversions::utf16_to_utf8(body_text);
        auto length = utf8body.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream<std::string>(
                std::move(utf8body)),
                length,
                std::move(content_type.append(::utility::conversions::to_utf16string("; charset=utf-8"))));
    }

    void set_body(const json::value &body_data)
    {
        auto body_text = utility::conversions::to_utf8string(body_data.serialize());
        auto length = body_text.size();
        set_body(concurrency::streams::bytestream::open_istream(std::move(body_text)), length, _XPLATSTR("application/json"));
    }

    void set_body(std::vector<unsigned char> &&body_data)
    {
        auto length = body_data.size();
        set_body(concurrency::streams::bytestream::open_istream(std::move(body_data)), length);
    }

    void set_body(const std::vector<unsigned char> &body_data)
    {
        set_body(concurrency::streams::bytestream::open_istream(body_data), body_data.size());
    }

    void set_body(const concurrency::streams::istream &stream, const utility::string_t &content_type = _XPLATSTR("application/octet-stream"))
    {
        _m_impl->set_body(stream, content_type);
    }

    void set_body(const concurrency::streams::istream &stream, utility::size64_t content_length, const utility::string_t &content_type = _XPLATSTR("application/octet-stream"))
    {
        _m_impl->set_body(stream, content_length, content_type);
    }

    concurrency::streams::istream body() const
    {
        return _m_impl->instream();
    }

    pplx::task<http::http_response> content_ready() const
    {
        http_response resp = *this;
        return pplx::create_task(_m_impl->_get_data_available()).then([resp](utility::size64_t) mutable { return resp; });
    }

    std::shared_ptr<http::details::_http_response> _get_impl() const { return _m_impl; }

    http::details::_http_server_context * _get_server_context() const { return _m_impl->_get_server_context(); }
    void _set_server_context(std::unique_ptr<http::details::_http_server_context> server_context) { _m_impl->_set_server_context(std::move(server_context)); }

private:

    std::shared_ptr<http::details::_http_response> _m_impl;
};

namespace details {
class _http_request : public http::details::http_msg_base, public std::enable_shared_from_this<_http_request>
{
public:

    _ASYNCRTIMP _http_request(http::method mtd);

    _ASYNCRTIMP _http_request(std::unique_ptr<http::details::_http_server_context> server_context);

    virtual ~_http_request() {}

    http::method &method() { return m_method; }

    uri &request_uri() { return m_uri; }

    _ASYNCRTIMP uri absolute_uri() const;

    _ASYNCRTIMP uri relative_uri() const;

    _ASYNCRTIMP void set_request_uri(const uri&);

    const pplx::cancellation_token &cancellation_token() const { return m_cancellationToken; }

    void set_cancellation_token(const pplx::cancellation_token &token)
    {
        m_cancellationToken = token;
    }

    _ASYNCRTIMP utility::string_t to_string() const;

    _ASYNCRTIMP pplx::task<void> reply(const http_response &response);

    pplx::task<http_response> get_response()
    {
        return pplx::task<http_response>(m_response);
    }

    _ASYNCRTIMP pplx::task<void> _reply_if_not_already(http::status_code status);

    void set_response_stream(const concurrency::streams::ostream &stream)
    {
        m_response_stream = stream;
    }

    void set_progress_handler(const progress_handler &handler)
    {
        m_progress_handler = std::make_shared<progress_handler>(handler);
    }

    const concurrency::streams::ostream & _response_stream() const { return m_response_stream; }

    const std::shared_ptr<progress_handler> & _progress_handler() const { return m_progress_handler; }

    http::details::_http_server_context * _get_server_context() const { return m_server_context.get(); }

    void _set_server_context(std::unique_ptr<http::details::_http_server_context> server_context) { m_server_context = std::move(server_context); }

    void _set_listener_path(const utility::string_t &path) { m_listener_path = path; }

    void _set_base_uri(const http::uri &base_uri) { m_base_uri = base_uri; }

private:

    // Actual initiates sending the response, without checking if a response has already been sent.
    pplx::task<void> _reply_impl(http_response response);

    http::method m_method;

    // Tracks whether or not a response has already been started for this message.
    pplx::details::atomic_long m_initiated_response;

    std::unique_ptr<http::details::_http_server_context> m_server_context;

    pplx::cancellation_token m_cancellationToken;

    http::uri m_base_uri;
    http::uri m_uri;
    utility::string_t m_listener_path;

    concurrency::streams::ostream m_response_stream;

    std::shared_ptr<progress_handler> m_progress_handler;

    pplx::task_completion_event<http_response> m_response;
};


}  // namespace details

class http_request
{
public:
    http_request()
        : _m_impl(std::make_shared<http::details::_http_request>(methods::GET)) {}

    http_request(http::method mtd)
        : _m_impl(std::make_shared<http::details::_http_request>(std::move(mtd))) {}

    ~http_request() {}

    const http::method &method() const { return _m_impl->method(); }

    void set_method(const http::method &method) const { _m_impl->method() = method; }

    const uri & request_uri() const { return _m_impl->request_uri(); }

    void set_request_uri(const uri& uri) { return _m_impl->set_request_uri(uri); }

    uri relative_uri() const { return _m_impl->relative_uri(); }

    uri absolute_uri() const { return _m_impl->absolute_uri(); }

    http_headers &headers() { return _m_impl->headers(); }

    const http_headers &headers() const { return _m_impl->headers(); }

    pplx::task<utility::string_t> extract_string(bool ignore_content_type = false)
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_string(ignore_content_type); });
    }

    pplx::task<utf8string> extract_utf8string(bool ignore_content_type = false)
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_utf8string(ignore_content_type); });
    }

    pplx::task<utf16string> extract_utf16string(bool ignore_content_type = false)
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->extract_utf16string(ignore_content_type); });
    }

    pplx::task<json::value> extract_json(bool ignore_content_type = false) const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl, ignore_content_type](utility::size64_t) { return impl->_extract_json(ignore_content_type); });
    }

    pplx::task<std::vector<unsigned char>> extract_vector() const
    {
        auto impl = _m_impl;
        return pplx::create_task(_m_impl->_get_data_available()).then([impl](utility::size64_t) { return impl->_extract_vector(); });
    }

    void set_body(utf8string &&body_text, const utf8string &content_type = utf8string("text/plain; charset=utf-8"))
    {
        const auto length = body_text.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream<std::string>(std::move(body_text)), length, content_type);
    }

    void set_body(const utf8string &body_text, const utf8string &content_type = utf8string("text/plain; charset=utf-8"))
    {
        _m_impl->set_body(concurrency::streams::bytestream::open_istream<std::string>(body_text), body_text.size(), content_type);
    }

    void set_body(const utf16string &body_text, utf16string content_type = ::utility::conversions::to_utf16string("text/plain"))
    {
        if(content_type.find(::utility::conversions::to_utf16string("charset=")) != content_type.npos)
        {
            throw std::invalid_argument("content_type can't contain a 'charset'.");
        }

        auto utf8body = utility::conversions::utf16_to_utf8(body_text);
        auto length = utf8body.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream(
                std::move(utf8body)),
                length,
                std::move(content_type.append(::utility::conversions::to_utf16string("; charset=utf-8"))));
    }

    void set_body(const json::value &body_data)
    {
        auto body_text = utility::conversions::to_utf8string(body_data.serialize());
        auto length = body_text.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream(std::move(body_text)), length, _XPLATSTR("application/json"));
    }

    void set_body(std::vector<unsigned char> &&body_data)
    {
        auto length = body_data.size();
        _m_impl->set_body(concurrency::streams::bytestream::open_istream(std::move(body_data)), length, _XPLATSTR("application/octet-stream"));
    }

    void set_body(const std::vector<unsigned char> &body_data)
    {
        set_body(concurrency::streams::bytestream::open_istream(body_data), body_data.size());
    }

    void set_body(const concurrency::streams::istream &stream, const utility::string_t &content_type = _XPLATSTR("application/octet-stream"))
    {
        _m_impl->set_body(stream, content_type);
    }

    void set_body(const concurrency::streams::istream &stream, utility::size64_t content_length, const utility::string_t &content_type = _XPLATSTR("application/octet-stream"))
    {
        _m_impl->set_body(stream, content_length, content_type);
    }

    concurrency::streams::istream body() const
    {
        return _m_impl->instream();
    }

    void set_response_stream(const concurrency::streams::ostream &stream)
    {
        return _m_impl->set_response_stream(stream);
    }

    void set_progress_handler(const progress_handler &handler)
    {
        return _m_impl->set_progress_handler(handler);
    }

    pplx::task<void> reply(const http_response &response) const { return _m_impl->reply(response); }

    pplx::task<void> reply(http::status_code status) const
    {
        return reply(http_response(status));
    }

    pplx::task<void> reply(http::status_code status, const json::value &body_data) const
    {
        http_response response(status);
        response.set_body(body_data);
        return reply(response);
    }

    //  Callers of this function do NOT need to block waiting for the response to be
    pplx::task<void> reply(http::status_code status, utf8string &&body_data, const utf8string &content_type = "text/plain; charset=utf-8") const
    {
        http_response response(status);
        response.set_body(std::move(body_data), content_type);
        return reply(response);
    }

    //  Callers of this function do NOT need to block waiting for the response to be
    pplx::task<void> reply(http::status_code status, const utf8string &body_data, const utf8string &content_type = "text/plain; charset=utf-8") const
    {
        http_response response(status);
        response.set_body(body_data, content_type);
        return reply(response);
    }

    //  Callers of this function do NOT need to block waiting for the response to be
    pplx::task<void> reply(http::status_code status, const utf16string &body_data, const utf16string &content_type = ::utility::conversions::to_utf16string("text/plain")) const
    {
        http_response response(status);
        response.set_body(body_data, content_type);
        return reply(response);
    }

    pplx::task<void> reply(status_code status, const concurrency::streams::istream &body, const utility::string_t &content_type = _XPLATSTR("application/octet-stream")) const
    {
        http_response response(status);
        response.set_body(body, content_type);
        return reply(response);
    }

    pplx::task<void> reply(status_code status, const concurrency::streams::istream &body, utility::size64_t content_length, const utility::string_t &content_type = _XPLATSTR("application/octet-stream")) const
    {
        http_response response(status);
        response.set_body(body, content_length, content_type);
        return reply(response);
    }

    pplx::task<http_request> content_ready() const
    {
        http_request req = *this;
        return pplx::create_task(_m_impl->_get_data_available()).then([req](utility::size64_t) mutable { return req; });
    }

    pplx::task<http_response> get_response() const
    {
        return _m_impl->get_response();
    }

    utility::string_t to_string() const { return _m_impl->to_string(); }

    pplx::task<void> _reply_if_not_already(status_code status) { return _m_impl->_reply_if_not_already(status); }

    http::details::_http_server_context * _get_server_context() const { return _m_impl->_get_server_context(); }

    static http_request _create_request(std::unique_ptr<http::details::_http_server_context> server_context) { return http_request(std::move(server_context)); }
    void _set_server_context(std::unique_ptr<http::details::_http_server_context> server_context) { _m_impl->_set_server_context(std::move(server_context)); }

    void _set_listener_path(const utility::string_t &path) { _m_impl->_set_listener_path(path); }

    const std::shared_ptr<http::details::_http_request> & _get_impl() const { return _m_impl; }

    void _set_cancellation_token(const pplx::cancellation_token &token)
    {
        _m_impl->set_cancellation_token(token);
    }

    const pplx::cancellation_token & _cancellation_token() const
    {
        return _m_impl->cancellation_token();
    }

    void _set_base_uri(const http::uri &base_uri)
    {
        _m_impl->_set_base_uri(base_uri);
    }

private:
    friend class http::details::_http_request;
    friend class http::client::http_client;

    http_request(std::unique_ptr<http::details::_http_server_context> server_context) : _m_impl(std::make_shared<details::_http_request>(std::move(server_context))) {}

    std::shared_ptr<http::details::_http_request> _m_impl;
};

class http_pipeline_stage : public std::enable_shared_from_this<http_pipeline_stage>
{
public:

    virtual ~http_pipeline_stage()
    {
    }

    virtual pplx::task<http_response> propagate(http_request request) = 0;

protected:

    http_pipeline_stage()
    {
    }

    const std::shared_ptr<http_pipeline_stage> & next_stage() const
    {
        return m_next_stage;
    }

    std::shared_ptr<http_pipeline_stage> current_stage()
    {
        return this->shared_from_this();
    }

private:
    friend class http_pipeline;

    void set_next_stage(const std::shared_ptr<http_pipeline_stage> &next)
    {
        m_next_stage = next;
    }

    std::shared_ptr<http_pipeline_stage> m_next_stage;

    // No copy or assignment.
    http_pipeline_stage & operator=(const http_pipeline_stage &);
    http_pipeline_stage(const http_pipeline_stage &);
};

namespace details {

class function_pipeline_wrapper : public http::http_pipeline_stage
{
public:
    function_pipeline_wrapper(std::function<pplx::task<http_response>(http_request, std::shared_ptr<http::http_pipeline_stage>)> handler) : m_handler(handler)
    {
    }

    virtual pplx::task<http_response> propagate(http_request request) override
    {
        return m_handler(request, next_stage());
    }
private:

    std::function<pplx::task<http_response>(http_request, std::shared_ptr<http::http_pipeline_stage>)> m_handler;
};

} // namespace details

class http_pipeline
{
public:

    static std::shared_ptr<http_pipeline> create_pipeline(const std::shared_ptr<http_pipeline_stage> &last)
    {
        return std::shared_ptr<http_pipeline>(new http_pipeline(last));
    }

    pplx::task<http_response> propagate(http_request request)
    {
        std::shared_ptr<http_pipeline_stage> first;
        {
            pplx::extensibility::scoped_recursive_lock_t l(m_lock);
            first = (m_stages.size() > 0) ? m_stages[0] : m_last_stage;
        }
        return first->propagate(request);
    }

    void append(const std::shared_ptr<http_pipeline_stage> &stage)
    {
        pplx::extensibility::scoped_recursive_lock_t l(m_lock);

        if (m_stages.size() > 0)
        {
            std::shared_ptr<http_pipeline_stage> penultimate = m_stages[m_stages.size()-1];
            penultimate->set_next_stage(stage);
        }
        stage->set_next_stage(m_last_stage);

        m_stages.push_back(stage);
    }

    void set_last_stage(const std::shared_ptr<http_pipeline_stage> &last)
    {
        m_last_stage = last;
    }

    const std::shared_ptr<http_pipeline_stage>& last_stage() const
    {
        return m_last_stage;
    }

private:

    http_pipeline(const std::shared_ptr<http_pipeline_stage> &last) : m_last_stage(last)
    {
    }

    // The vector of pipeline stages.
    std::vector<std::shared_ptr<http_pipeline_stage>> m_stages;

    // The last stage is always set up by the client or listener and cannot
    // be changed. All application-defined stages are executed before the
    // last stage, which is typically a send or dispatch.
    std::shared_ptr<http_pipeline_stage> m_last_stage;

    pplx::extensibility::recursive_lock_t m_lock;

    // No copy or assignment.
    http_pipeline & operator=(const http_pipeline &);
    http_pipeline(const http_pipeline &);
};

}}