Add WSClient

Builds/VisualStudio2015/RippleD.vcxproj

@@ -3431,6 +3431,14 @@
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug|x64'">True</ExcludedFromBuild>
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='release|x64'">True</ExcludedFromBuild>
     </ClCompile>
+    <ClCompile Include="..\..\src\ripple\test\impl\WSClient.cpp">
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug|x64'">True</ExcludedFromBuild>
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='release|x64'">True</ExcludedFromBuild>
+    </ClCompile>
+    <ClCompile Include="..\..\src\ripple\test\impl\WSClient_test.cpp">
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug|x64'">True</ExcludedFromBuild>
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='release|x64'">True</ExcludedFromBuild>
+    </ClCompile>
     <ClInclude Include="..\..\src\ripple\test\JSONRPCClient.h">
     </ClInclude>
     <ClInclude Include="..\..\src\ripple\test\jtx.h">
@@ -3607,6 +3615,8 @@
     </ClCompile>
     <ClInclude Include="..\..\src\ripple\test\mao\Net.h">
     </ClInclude>
+    <ClInclude Include="..\..\src\ripple\test\WSClient.h">
+    </ClInclude>
     <ClCompile Include="..\..\src\ripple\unity\app_ledger.cpp">
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug.classic|x64'">True</ExcludedFromBuild>
       <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='release.classic|x64'">True</ExcludedFromBuild>
@@ -3765,6 +3775,10 @@
     </ClCompile>
     <ClInclude Include="..\..\src\ripple\websocket\WebSocket02.h">
     </ClInclude>
+    <ClInclude Include="..\..\src\ripple\wsproto\basic_socket.h">
+    </ClInclude>
+    <ClInclude Include="..\..\src\ripple\wsproto\wsproto.h">
+    </ClInclude>
     <ClCompile Include="..\..\src\rocksdb2\db\builder.cc">
       <ExcludedFromBuild>True</ExcludedFromBuild>
     </ClCompile>

Builds/VisualStudio2015/RippleD.vcxproj.filters

@@ -406,6 +406,9 @@
     <Filter Include="ripple\websocket">
       <UniqueIdentifier>{44780F86-42D3-2F2B-0846-5AEE2CA6D7FE}</UniqueIdentifier>
     </Filter>
+    <Filter Include="ripple\wsproto">
+      <UniqueIdentifier>{1D54E820-ADC9-94FB-19E7-653EFDE4CBE9}</UniqueIdentifier>
+    </Filter>
     <Filter Include="rocksdb2">
       <UniqueIdentifier>{15B4B65A-0F03-7BA9-38CD-42A5712392CB}</UniqueIdentifier>
     </Filter>
@@ -3924,6 +3927,12 @@
     <ClCompile Include="..\..\src\ripple\test\impl\ManualTimeKeeper.cpp">
       <Filter>ripple\test\impl</Filter>
     </ClCompile>
+    <ClCompile Include="..\..\src\ripple\test\impl\WSClient.cpp">
+      <Filter>ripple\test\impl</Filter>
+    </ClCompile>
+    <ClCompile Include="..\..\src\ripple\test\impl\WSClient_test.cpp">
+      <Filter>ripple\test\impl</Filter>
+    </ClCompile>
     <ClInclude Include="..\..\src\ripple\test\JSONRPCClient.h">
       <Filter>ripple\test</Filter>
     </ClInclude>
@@ -4110,6 +4119,9 @@
     <ClInclude Include="..\..\src\ripple\test\mao\Net.h">
       <Filter>ripple\test\mao</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\src\ripple\test\WSClient.h">
+      <Filter>ripple\test</Filter>
+    </ClInclude>
     <ClCompile Include="..\..\src\ripple\unity\app_ledger.cpp">
       <Filter>ripple\unity</Filter>
     </ClCompile>
@@ -4245,6 +4257,12 @@
     <ClInclude Include="..\..\src\ripple\websocket\WebSocket02.h">
       <Filter>ripple\websocket</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\src\ripple\wsproto\basic_socket.h">
+      <Filter>ripple\wsproto</Filter>
+    </ClInclude>
+    <ClInclude Include="..\..\src\ripple\wsproto\wsproto.h">
+      <Filter>ripple\wsproto</Filter>
+    </ClInclude>
     <ClCompile Include="..\..\src\rocksdb2\db\builder.cc">
       <Filter>rocksdb2\db</Filter>
     </ClCompile>

src/ripple/test/WSClient.h

@@ -0,0 +1,55 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2016 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_TEST_WSCLIENT_H_INCLUDED
+#define RIPPLE_TEST_WSCLIENT_H_INCLUDED
+
+#include <ripple/test/AbstractClient.h>
+#include <ripple/core/Config.h>
+#include <boost/optional.hpp>
+#include <chrono>
+#include <memory>
+
+namespace ripple {
+namespace test {
+
+class WSClient : public AbstractClient
+{
+public:
+    /** Retrieve a message. */
+    virtual
+    boost::optional<Json::Value>
+    getMsg(std::chrono::milliseconds const& timeout =
+        std::chrono::milliseconds{0}) = 0;
+
+    /** Retrieve a message that meets the predicate criteria. */
+    virtual
+    boost::optional<Json::Value>
+    findMsg(std::chrono::milliseconds const& timeout,
+        std::function<bool(Json::Value const&)> pred) = 0;
+};
+
+/** Returns a client operating through WebSockets/S. */
+std::unique_ptr<WSClient>
+makeWSClient(Config const& cfg);
+
+} // test
+} // ripple
+
+#endif

src/ripple/test/impl/WSClient.cpp

@@ -0,0 +1,313 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2016 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <BeastConfig.h>
+#include <ripple/test/WSClient.h>
+#include <ripple/test/jtx.h>
+#include <ripple/json/json_reader.h>
+#include <ripple/json/to_string.h>
+#include <ripple/protocol/JsonFields.h>
+#include <ripple/server/Port.h>
+#include <ripple/wsproto/wsproto.h>
+#include <condition_variable>
+
+#include <beast/unit_test/suite.h>
+
+namespace ripple {
+namespace test {
+
+class WSClientImpl : public WSClient
+{
+    using error_code = boost::system::error_code;
+
+    struct msg
+    {
+        Json::Value jv;
+
+        explicit
+        msg(Json::Value&& jv_)
+            : jv(jv_)
+        {
+        }
+    };
+
+    class read_frame_op
+    {
+        struct data
+        {
+            WSClientImpl& wsc;
+            wsproto::frame_header fh;
+            boost::asio::streambuf sb;
+
+            data(WSClientImpl& wsc_)
+                : wsc(wsc_)
+            {
+            }
+
+            ~data()
+            {
+                wsc.on_read_done();
+            }
+        };
+
+        std::shared_ptr<data> d_;
+
+    public:
+        read_frame_op(read_frame_op const&) = default;
+        read_frame_op(read_frame_op&&) = default;
+
+        explicit
+        read_frame_op(WSClientImpl& wsc)
+            : d_(std::make_shared<data>(wsc))
+        {
+            read_one();
+        }
+
+        void read_one()
+        {
+            // hack
+            d_->sb.consume(d_->sb.size());
+            d_->wsc.ws_.async_read_fh(
+                d_->fh, std::move(*this));
+        }
+
+        void operator()(error_code const& ec)
+        {
+            if(ec)
+                return d_->wsc.on_read_frame(
+                    ec, d_->fh, 0, d_->sb.data(),
+                        std::move(*this));
+            d_->wsc.ws_.async_read(d_->fh,
+                d_->sb.prepare(d_->fh.len), std::move(*this));
+        }
+
+        void operator()(error_code const& ec,
+            wsproto::frame_header const& fh,
+                std::size_t bytes_transferred)
+        {
+            if(ec)
+                return d_->wsc.on_read_frame(
+                    ec, d_->fh, 0, d_->sb.data(),
+                        std::move(*this));
+            if(d_->fh.mask)
+            {
+                // TODO: apply key mask to payload
+            }
+            d_->sb.commit(bytes_transferred);
+            return d_->wsc.on_read_frame(
+                ec, d_->fh, bytes_transferred, d_->sb.data(),
+                    std::move(*this));
+       }
+    };
+
+    static
+    boost::asio::ip::tcp::endpoint
+    getEndpoint(BasicConfig const& cfg)
+    {
+        auto& log = std::cerr;
+        ParsedPort common;
+        parse_Port (common, cfg["server"], log);
+        for (auto const& name : cfg.section("server").values())
+        {
+            if (! cfg.exists(name))
+                continue;
+            ParsedPort pp;
+            parse_Port(pp, cfg[name], log);
+            if(pp.protocol.count("ws") == 0)
+                continue;
+            using boost::asio::ip::address_v4;
+            if(*pp.ip == address_v4{0x00000000})
+                *pp.ip = address_v4{0x7f000001};
+            return { *pp.ip, *pp.port };
+        }
+        throw std::runtime_error("Missing WebSocket port");
+    }
+
+    template <class ConstBuffers>
+    static
+    std::string
+    buffer_string (ConstBuffers const& b)
+    {
+        using namespace boost::asio;
+        std::string s;
+        s.resize(buffer_size(b));
+        buffer_copy(buffer(&s[0], s.size()), b);
+        return s;
+    }
+
+    boost::asio::io_service ios_;
+    boost::optional<
+        boost::asio::io_service::work> work_;
+    std::thread thread_;
+    boost::asio::ip::tcp::socket stream_;
+    wsproto::basic_socket<boost::asio::ip::tcp::socket&> ws_;
+
+    // synchronize destructor
+    bool b0_ = false;
+    std::mutex m0_;
+    std::condition_variable cv0_;
+
+    // sychronize message queue
+    std::mutex m_;
+    std::condition_variable cv_;
+    std::list<std::shared_ptr<msg>> msgs_;
+
+public:
+    explicit
+    WSClientImpl(Config const& cfg)
+        : work_(ios_)
+        , thread_([&]{ ios_.run(); })
+        , stream_(ios_)
+        , ws_(stream_)
+    {
+        using namespace boost::asio;
+        stream_.connect(getEndpoint(cfg));
+        error_code ec;
+        ws_.connect(ec);
+        if(ec)
+            throw ec;
+        read_frame_op{*this};
+    }
+
+    ~WSClientImpl() override
+    {
+        stream_.close();
+        {
+            std::unique_lock<std::mutex> lock(m0_);
+            cv0_.wait(lock, [&]{ return b0_; });
+        }
+        work_ = boost::none;
+        thread_.join();
+
+        //stream_.shutdown(boost::asio::ip::tcp::socket::shutdown_both);
+        //stream_.close();
+    }
+
+    Json::Value
+    invoke(std::string const& cmd,
+        Json::Value const& params) override
+    {
+        using namespace boost::asio;
+        using namespace std::chrono_literals;
+
+        {
+            Json::Value jp;
+            if(params)
+               jp = params;
+            jp["command"] = cmd;
+            auto const s = to_string(jp);
+            ws_.write(buffer(s));
+        }
+
+        auto jv = findMsg(5s,
+            [&](Json::Value const& jv)
+            {
+                return jv[jss::type] == jss::response;
+            });
+        if (jv)
+            return *jv;
+        return {};
+    }
+
+    boost::optional<Json::Value>
+    getMsg(std::chrono::milliseconds const& timeout) override
+    {
+        std::shared_ptr<msg> m;
+        {
+            std::unique_lock<std::mutex> lock(m_);
+            if(! cv_.wait_for(lock, timeout,
+                    [&]{ return ! msgs_.empty(); }))
+                return boost::none;
+            m = std::move(msgs_.back());
+            msgs_.pop_back();
+        }
+        return std::move(m->jv);
+    }
+
+    boost::optional<Json::Value>
+    findMsg(std::chrono::milliseconds const& timeout,
+        std::function<bool(Json::Value const&)> pred) override
+    {
+        std::shared_ptr<msg> m;
+        {
+            std::unique_lock<std::mutex> lock(m_);
+            if(! cv_.wait_for(lock, timeout,
+                [&]
+                {
+                    for (auto it = msgs_.begin();
+                        it != msgs_.end(); ++it)
+                    {
+                        if (pred((*it)->jv))
+                        {
+                            m = std::move(*it);
+                            msgs_.erase(it);
+                            return true;
+                        }
+                    }
+                    return false;
+                }))
+            {
+                return boost::none;
+            }
+        }
+        return std::move(m->jv);
+    }
+
+private:
+    template<class ConstBuffers>
+    void
+    on_read_frame(error_code const& ec,
+        wsproto::frame_header const& fh,
+            std::size_t bytes_transferred,
+                ConstBuffers const& b,
+                    read_frame_op&& op)
+    {
+        if(bytes_transferred == 0)
+            return;
+        Json::Value jv;
+        Json::Reader jr;
+        jr.parse(buffer_string(b), jv);
+        auto m = std::make_shared<msg>(
+            std::move(jv));
+        {
+            std::lock_guard<std::mutex> lock(m_);
+            msgs_.push_front(m);
+            cv_.notify_all();
+        }
+        op.read_one();
+    }
+
+    // Called when the read op terminates
+    void
+    on_read_done()
+    {
+        std::lock_guard<std::mutex> lock(m_);
+        b0_ = true;
+        cv0_.notify_all();
+    }
+};
+
+std::unique_ptr<WSClient>
+makeWSClient(Config const& cfg)
+{
+    return std::make_unique<WSClientImpl>(cfg);
+}
+
+} // test
+} // ripple

src/ripple/test/impl/WSClient_test.cpp

@@ -0,0 +1,63 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2016 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include <BeastConfig.h>
+#include <ripple/test/WSClient.h>
+#include <ripple/test/jtx.h>
+#include <beast/unit_test/suite.h>
+
+namespace ripple {
+namespace test {
+
+class WSClient_test : public beast::unit_test::suite
+{
+public:
+    void run() override
+    {
+        using namespace jtx;
+        Env env(*this);
+        auto wsc = makeWSClient(env.app().config());
+        {
+            Json::Value jv;
+            jv["streams"] = Json::arrayValue;
+            jv["streams"].append("ledger");
+            //jv["streams"].append("server");
+            //jv["streams"].append("transactions");
+            //jv["streams"].append("transactions_proposed");
+            log << pretty(wsc->invoke("subscribe", jv));
+        }
+        env.fund(XRP(10000), "alice");
+        env.close();
+        /*
+        env.fund(XRP(10000), "dan", "eric", "fred");
+        env.close();
+        env.fund(XRP(10000), "george", "harold", "iris");
+        env.close();
+        */
+        auto jv = wsc->getMsg(std::chrono::seconds(1));
+        if(jv)
+            log << pretty(*jv);
+        pass();
+    }
+};
+
+BEAST_DEFINE_TESTSUITE(WSClient,test,ripple);
+
+} // test
+} // ripple

src/ripple/unity/test.cpp

@@ -50,3 +50,5 @@
 #include <ripple/test/impl/BasicNetwork_test.cpp>
 #include <ripple/test/impl/JSONRPCClient.cpp>
 #include <ripple/test/impl/ManualTimeKeeper.cpp>
+#include <ripple/test/impl/WSClient.cpp>
+#include <ripple/test/impl/WSClient_test.cpp>