ws_client.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.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* Websocket client side implementation
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#ifndef _CASA_WS_CLIENT_H
#define _CASA_WS_CLIENT_H

#include "cpprest/details/basic_types.h"

#if !defined(CPPREST_EXCLUDE_WEBSOCKETS)

#include <memory>
#include <limits>
#include <condition_variable>
#include <mutex>

#include "pplx/pplxtasks.h"
#include "cpprest/uri.h"
#include "cpprest/details/web_utilities.h"
#include "cpprest/http_headers.h"
#include "cpprest/asyncrt_utils.h"
#include "cpprest/ws_msg.h"

namespace web
{

// For backwards compatibility for when in the experimental namespace.
// At next major release this should be deleted.
namespace experimental = web;

// In the past namespace was accidentally called 'web_sockets'. To avoid breaking code
// alias it. At our next major release this should be deleted.
namespace web_sockets = websockets;

namespace websockets
{
namespace client
{

enum class websocket_close_status
{
    normal = 1000,
    going_away = 1001,
    protocol_error = 1002,
    unsupported = 1003, //or data_mismatch
    abnormal_close = 1006,
    inconsistent_datatype = 1007,
    policy_violation = 1008,
    too_large = 1009,
    negotiate_error = 1010,
    server_terminate = 1011,
};

class websocket_client_config
{
public:

    websocket_client_config() : m_sni_enabled(true) {}

    const web_proxy& proxy() const
    {
        return m_proxy;
    }

    void set_proxy(const web_proxy &proxy)
    {
        m_proxy = proxy;
    }

    const web::credentials& credentials() const
    {
        return m_credentials;
    }

    void set_credentials(const web::credentials &cred)
    {
        m_credentials = cred;
    }

    void disable_sni()
    {
        m_sni_enabled = false;
    }

    bool is_sni_enabled() const
    {
        return m_sni_enabled;
    }

    void set_server_name(const utf8string &name)
    {
        m_sni_hostname = name;
    }

    const utf8string & server_name() const
    {
        return m_sni_hostname;
    }

    web::http::http_headers &headers() { return m_headers; }

    const web::http::http_headers &headers() const { return m_headers; }

    _ASYNCRTIMP void add_subprotocol(const ::utility::string_t &name);

    _ASYNCRTIMP std::vector<::utility::string_t> subprotocols() const;

private:
    web::web_proxy m_proxy;
    web::credentials m_credentials;
    web::http::http_headers m_headers;
    bool m_sni_enabled;
    utf8string m_sni_hostname;
};

class websocket_exception : public std::exception
{
public:

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

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

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

    websocket_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
    websocket_exception(int errorCode, std::string whatArg)
        : m_errorCode(utility::details::create_error_code(errorCode)),
        m_msg(std::move(whatArg))
    {}

    websocket_exception(std::error_code code, std::string whatArg) :
        m_errorCode(std::move(code)),
        m_msg(std::move(whatArg))
    {}
#endif

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

    websocket_exception(std::error_code code, const utility::string_t &whatArg) :
        m_errorCode(std::move(code)),
        m_msg(utility::conversions::to_utf8string(whatArg))
    {}

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

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

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

namespace details
{

// Interface to be implemented by the websocket client callback implementations.
class websocket_client_callback_impl
{
public:

    websocket_client_callback_impl(websocket_client_config config) :
        m_config(std::move(config)) {}

    virtual ~websocket_client_callback_impl() CPPREST_NOEXCEPT{}

    virtual pplx::task<void> connect() = 0;

    virtual pplx::task<void> send(websocket_outgoing_message &msg) = 0;

    virtual void set_message_handler(const std::function<void(const websocket_incoming_message&)>& handler) = 0;

    virtual pplx::task<void> close() = 0;

    virtual pplx::task<void> close(websocket_close_status close_status, const utility::string_t &close_reason = _XPLATSTR("")) = 0;

    virtual void set_close_handler(const std::function<void(websocket_close_status, const utility::string_t&, const std::error_code&)>& handler) = 0;

    const web::uri& uri() const
    {
        return m_uri;
    }

    void set_uri(const web::uri &uri)
    {
        m_uri = uri;
    }

    const websocket_client_config& config() const
    {
        return m_config;
    }

    static void verify_uri(const web::uri& uri)
    {
        // Most of the URI schema validation is taken care by URI class.
        // We only need to check certain things specific to websockets.
        if (uri.scheme() != _XPLATSTR("ws") && uri.scheme() != _XPLATSTR("wss"))
        {
            throw std::invalid_argument("URI scheme must be 'ws' or 'wss'");
        }

        if (uri.host().empty())
        {
            throw std::invalid_argument("URI must contain a hostname.");
        }

        // Fragment identifiers are meaningless in the context of WebSocket URIs
        // and MUST NOT be used on these URIs.
        if (!uri.fragment().empty())
        {
            throw std::invalid_argument("WebSocket URI must not contain fragment identifiers");
        }
    }

protected:
    web::uri m_uri;
    websocket_client_config m_config;
};

// Interface to be implemented by the websocket client task implementations.
class websocket_client_task_impl
{

public:
    _ASYNCRTIMP websocket_client_task_impl(websocket_client_config config);

    _ASYNCRTIMP virtual ~websocket_client_task_impl() CPPREST_NOEXCEPT;

    _ASYNCRTIMP pplx::task<websocket_incoming_message> receive();

    _ASYNCRTIMP void close_pending_tasks_with_error(const websocket_exception &exc);

    const std::shared_ptr<websocket_client_callback_impl> & callback_client() const { return m_callback_client; };

private:
    void set_handler();

    // When a message arrives, if there are tasks waiting for a message, signal the topmost one.
    // Else enqueue the message in a queue.
    // m_receive_queue_lock : to guard access to the queue & m_client_closed
    std::mutex m_receive_queue_lock;
    // Queue to store incoming messages when there are no tasks waiting for a message
    std::queue<websocket_incoming_message> m_receive_msg_queue;
    // Queue to maintain the receive tasks when there are no messages(yet).
    std::queue<pplx::task_completion_event<websocket_incoming_message>> m_receive_task_queue;

    // Initially set to false, becomes true if a close frame is received from the server or
    // if the underlying connection is aborted or terminated.
    bool m_client_closed;

    std::shared_ptr<websocket_client_callback_impl> m_callback_client;
};
}

class websocket_client
{
public:
    websocket_client() :
        m_client(std::make_shared<details::websocket_client_task_impl>(websocket_client_config()))
    {}

    websocket_client(websocket_client_config config) :
        m_client(std::make_shared<details::websocket_client_task_impl>(std::move(config)))
    {}

    pplx::task<void> connect(const web::uri &uri)
    {
        m_client->callback_client()->verify_uri(uri);
        m_client->callback_client()->set_uri(uri);
        auto client = m_client;
        return m_client->callback_client()->connect().then([client](pplx::task<void> result)
        {
            try
            {
                result.get();
            }
            catch (const websocket_exception& ex)
            {
                client->close_pending_tasks_with_error(ex);
                throw;
            }
        });
    }

    pplx::task<void> send(websocket_outgoing_message msg)
    {
        return m_client->callback_client()->send(msg);
    }

    pplx::task<websocket_incoming_message> receive()
    {
        return m_client->receive();
    }

    pplx::task<void> close()
    {
        return m_client->callback_client()->close();
    }

    pplx::task<void> close(websocket_close_status close_status, const utility::string_t& close_reason=_XPLATSTR(""))
    {
        return m_client->callback_client()->close(close_status, close_reason);
    }

    const web::uri& uri() const
    {
        return m_client->callback_client()->uri();
    }

    const websocket_client_config& config() const
    {
        return m_client->callback_client()->config();
    }

private:
    std::shared_ptr<details::websocket_client_task_impl> m_client;
};

class websocket_callback_client
{
public:
    _ASYNCRTIMP websocket_callback_client();

    _ASYNCRTIMP websocket_callback_client(websocket_client_config client_config);

    pplx::task<void> connect(const web::uri &uri)
    {
        m_client->verify_uri(uri);
        m_client->set_uri(uri);
        return m_client->connect();
    }

    pplx::task<void> send(websocket_outgoing_message msg)
    {
        return m_client->send(msg);
    }

    void set_message_handler(const std::function<void(const websocket_incoming_message& msg)>& handler)
    {
        m_client->set_message_handler(handler);
    }

    pplx::task<void> close()
    {
        return m_client->close();
    }

    pplx::task<void> close(websocket_close_status close_status, const utility::string_t& close_reason = _XPLATSTR(""))
    {
        return m_client->close(close_status, close_reason);
    }

    void set_close_handler(const std::function<void(websocket_close_status close_status, const utility::string_t& reason, const std::error_code& error)>& handler)
    {
        m_client->set_close_handler(handler);
    }

    const web::uri& uri() const
    {
        return m_client->uri();
    }

    const websocket_client_config& config() const
    {
        return m_client->config();
    }

private:
    std::shared_ptr<details::websocket_client_callback_impl> m_client;
};

}}}

#endif

#endif