fixed client protocol low level API (#49)

* fixed client protocol low level API: enables usage of providing clients along with other websocket protocols

* fix: made some c'tors explicit

cmake/unit_tests.cmake

@@ -11,6 +11,8 @@ add_executable(alltests
 	test/file_utils.cc
 	test/msleep.cc
 	test/die_if.cc
+	test/timeout_watcher.cc
+	test/fake_adapter_server.cc
 	test/mock_authenticator.cc
 	test/mock_request.cc
 	test/core/test_container_of.cc
@@ -42,6 +44,7 @@ add_executable(alltests
 	test/adapter/jsonrpc/test_proxy.cc
 	test/provider/test_url.cc
 	test/provider/test_static_filesystem.cc
+	test/provider/test_client_protocol.cc
 	test/integration/test_integration.cc
 	test/integration/server.cc
 	test/integration/provider.cc

include/webfuse/provider/client_protocol.h

@@ -24,34 +24,36 @@ extern "C"
 //------------------------------------------------------------------------------
 struct wfp_client_protocol;
 
-//------------------------------------------------------------------------------
-/// \struct wfp_provider
-/// \brief Provider.
-///
-/// \todo How is a user supposed to get a provider's instance?
-//------------------------------------------------------------------------------
-struct wfp_provider;
-
 //------------------------------------------------------------------------------
 /// \struct lws_protocols
 /// \brief Forward declaration of libwebsockets protocols structure.
 //------------------------------------------------------------------------------
 struct lws_protocols;
 
+//------------------------------------------------------------------------------
+/// \struct lws_context
+/// \brief Forward declaration of libwebsockets context structure.
+//------------------------------------------------------------------------------
+struct lws_context;
+
+//------------------------------------------------------------------------------
+/// \struct wfp_client_config
+/// \copydoc wfp_client_config
+//------------------------------------------------------------------------------
+struct wfp_client_config;
+
 //------------------------------------------------------------------------------
 /// \brief Creates a new webfuse provider client protocol.
 ///
-/// \note The user is responsible to manage lifetime of \arg user_data.
+/// \note The user is responsible to manage lifetime of \arg config.
 ///
-/// \todo How is a user supposed to get a provider's instance?
+/// \note TLS configuration is ignored, since TLS is managed by libwebsockets.
 ///
-/// \param provider pointer to provider
+/// \param config pointer to client config
-/// \param user_data user defined context
 /// \return newly created protocol
 //------------------------------------------------------------------------------
 extern WFP_API struct wfp_client_protocol * wfp_client_protocol_create(
-    struct wfp_provider const * provider,
+    struct wfp_client_config const * config);
-    void * user_data);
 
 //------------------------------------------------------------------------------
 /// \brief Disposes a protocol.
@@ -73,6 +75,26 @@ extern WFP_API void wfp_client_protocol_init_lws(
     struct wfp_client_protocol * protocol,
     struct lws_protocols * lws_protocol);
 
+
+//------------------------------------------------------------------------------
+/// \brief Connects the protocol to a remote webfuse adapter server.
+///
+/// \note This call starts to establish a connection. A callback is invoked,
+///       when the connection is estanlished.
+///
+/// \param protocol pointer to protocol
+/// \param context lws context
+/// \param url URL of remote webfuse adapter server
+///
+/// \see wfp_connected_fn
+/// \see wfp_client_config_set_onconnected
+//------------------------------------------------------------------------------
+//------------------------------------------------------------------------------
+extern WFP_API void wfp_client_protocol_connect(
+    struct wfp_client_protocol * protocol,
+    struct lws_context * context,
+    char const * url);
+
 #ifdef __cplusplus
 }
 #endif

lib/webfuse/provider/api.c

@@ -160,10 +160,9 @@ void wfp_client_config_set_onread(
 
 
 struct wfp_client_protocol * wfp_client_protocol_create(
-    struct wfp_provider const * provider,
+    struct wfp_client_config const * config)
-    void * user_data)
 {
-    return wfp_impl_client_protocol_create(provider, user_data);
+    return wfp_impl_client_protocol_create(config);
 }
 
 void wfp_client_protocol_dispose(
@@ -179,6 +178,15 @@ void wfp_client_protocol_init_lws(
     wfp_impl_client_protocol_init_lws(protocol, lws_protocol);
 }
 
+void wfp_client_protocol_connect(
+    struct wfp_client_protocol * protocol,
+    struct lws_context * context,
+    char const * url)
+{
+    wfp_impl_client_protocol_connect(protocol, context, url);
+}
+
+
 // client
 
 struct wfp_client * wfp_client_create(

lib/webfuse/provider/impl/client.c

@@ -9,10 +9,8 @@
 #include "webfuse/provider/impl/provider.h"
 #include "webfuse/provider/impl/client_protocol.h"
 #include "webfuse/provider/impl/client_config.h"
-#include "webfuse/provider/impl/url.h"
 #include "webfuse/core/lws_log.h"
 
-#define WFP_PROTOCOL ("fs")
 #define WFP_CLIENT_PROTOCOL_COUNT 2
 
 struct wfp_client
@@ -37,7 +35,7 @@ struct wfp_client * wfp_impl_client_create(
         wfp_impl_client_protocol_init(&client->protocol, &config->provider, config->user_data);
 
         memset(client->protocols, 0, sizeof(struct lws_protocols) * WFP_CLIENT_PROTOCOL_COUNT);
-        client->protocols[0].name = "fs";
+        client->protocols[0].name = WFP_CLIENT_PROTOCOL_NAME;
         wfp_impl_client_protocol_init_lws(&client->protocol, &client->protocols[0]);
 
         memset(&client->info, 0, sizeof(struct lws_context_creation_info));
@@ -69,26 +67,7 @@ void wfp_impl_client_connect(
     struct wfp_client * client,
     char const * url)
 {
-    struct wfp_impl_url url_data;
+    wfp_impl_client_protocol_connect(&client->protocol, client->context, url);
-    bool const success = wfp_impl_url_init(&url_data, url);
-    if (success)
-    {
-        struct lws_client_connect_info info;
-        memset(&info, 0, sizeof(struct lws_client_connect_info));
-        info.context = client->context;
-        info.port = url_data.port;
-        info.address = url_data.host;
-        info.path = url_data.path;
-        info.host = info.address;
-        info.origin = info.address;
-        info.ssl_connection = (url_data.use_tls) ? LCCSCF_USE_SSL : 0;
-        info.protocol = WFP_PROTOCOL;
-        info.pwsi = &client->protocol.wsi;
-
-        lws_client_connect_via_info(&info);
-
-        wfp_impl_url_cleanup(&url_data);
-    }
 }
 
 void wfp_impl_client_disconnect(

lib/webfuse/provider/impl/client_protocol.c

@@ -7,11 +7,13 @@
 #include <jansson.h>
 
 
+#include "webfuse/provider/impl/client_config.h"
 #include "webfuse/provider/impl/provider.h"
 #include "webfuse/core/util.h"
 #include "webfuse/core/message.h"
 #include "webfuse/core/message_queue.h"
 #include "webfuse/core/container_of.h"
+#include "webfuse/provider/impl/url.h"
 
 static void wfp_impl_client_protocol_respond(
     json_t * response,
@@ -156,13 +158,12 @@ void wfp_impl_client_protocol_cleanup(
 }
 
 struct wfp_client_protocol * wfp_impl_client_protocol_create(
-    struct wfp_provider const * provider,
+    struct wfp_client_config const * config)
-    void * user_data)
 {
     struct wfp_client_protocol * protocol = malloc(sizeof(struct wfp_client_protocol));
     if (NULL != protocol)
     {
-        wfp_impl_client_protocol_init(protocol, provider, user_data);
+        wfp_impl_client_protocol_init(protocol, &config->provider, config->user_data);
     }
 
     return protocol;
@@ -183,3 +184,35 @@ void wfp_impl_client_protocol_init_lws(
 	lws_protocol->per_session_data_size = 0;
 	lws_protocol->user = protocol;
 }
+
+void wfp_impl_client_protocol_connect(
+    struct wfp_client_protocol * protocol,
+    struct lws_context * context,
+    char const * url)
+{
+    struct wfp_impl_url url_data;
+    bool const success = wfp_impl_url_init(&url_data, url);
+    if (success)
+    {
+        struct lws_client_connect_info info;
+        memset(&info, 0, sizeof(struct lws_client_connect_info));
+        info.context = context;
+        info.port = url_data.port;
+        info.address = url_data.host;
+        info.path = url_data.path;
+        info.host = info.address;
+        info.origin = info.address;
+        info.ssl_connection = (url_data.use_tls) ? LCCSCF_USE_SSL : 0;
+        info.protocol = WFP_CLIENT_PROTOCOL_NAME;
+        info.pwsi = &protocol->wsi;
+
+        lws_client_connect_via_info(&info);
+
+        wfp_impl_url_cleanup(&url_data);
+    }
+    else
+    {
+        protocol->provider.disconnected(protocol->user_data);
+    }
+    
+}

lib/webfuse/provider/impl/client_protocol.h

@@ -11,8 +11,11 @@ extern "C"
 {
 #endif
 
-struct wfp_provider;
+#define WFP_CLIENT_PROTOCOL_NAME ("fs")
+
+struct wfp_client_config;
 struct lws_protocols;
+struct lws_context;
 
 struct wfp_client_protocol
 {
@@ -33,8 +36,7 @@ extern void wfp_impl_client_protocol_cleanup(
     struct wfp_client_protocol * protocol);
 
 extern struct wfp_client_protocol * wfp_impl_client_protocol_create(
-    struct wfp_provider const * provider,
+    struct wfp_client_config const * config);
-    void * user_data);
 
 extern void wfp_impl_client_protocol_dispose(
     struct wfp_client_protocol * protocol);
@@ -43,6 +45,11 @@ extern void wfp_impl_client_protocol_init_lws(
     struct wfp_client_protocol * protocol,
     struct lws_protocols * lws_protocol);
 
+extern void wfp_impl_client_protocol_connect(
+    struct wfp_client_protocol * protocol,
+    struct lws_context * context,
+    char const * url);
+
 #ifdef __cplusplus
 } 
 #endif

test/fake_adapter_server.cc

@@ -0,0 +1,196 @@
+#include "fake_adapter_server.hpp"
+#include "timeout_watcher.hpp"
+
+#include "webfuse/core/util.h"
+#include <libwebsockets.h>
+#include <cstring>
+#include <vector>
+#include <queue>
+#include <string>
+
+using webfuse_test::TimeoutWatcher;
+
+#define DEFAULT_TIMEOUT (std::chrono::milliseconds(5 * 1000))
+
+namespace
+{
+
+class IServer
+{
+public:
+    virtual ~IServer() = default;
+    virtual void onConnectionEstablished(struct lws * wsi) = 0;
+    virtual void onConnectionClosed(struct lws * wsi) = 0;
+    virtual void onMessageReceived(struct lws * wsi, char const * data, size_t length) = 0;
+    virtual void onWritable(struct lws * wsi) = 0;
+};
+
+}
+
+extern "C"
+{
+
+static int wf_test_fake_adapter_server_callback(
+	struct lws * wsi,
+	enum lws_callback_reasons reason,
+	void * WF_UNUSED_PARAM(user),
+	void * in,
+	size_t len)
+{
+    struct lws_protocols const * ws_protocol = lws_get_protocol(wsi);
+    if (NULL == ws_protocol)
+    {
+        return 0;
+    }
+
+    auto * server = reinterpret_cast<IServer*>(ws_protocol->user);
+    switch(reason)
+    {
+		case LWS_CALLBACK_ESTABLISHED:
+            server->onConnectionEstablished(wsi);
+            break;
+        case LWS_CALLBACK_CLOSED:
+            server->onConnectionClosed(wsi);
+            break;
+        case LWS_CALLBACK_RECEIVE:
+            {
+                auto * data = reinterpret_cast<char const *>(in);
+                server->onMessageReceived(wsi, data, len);
+            }
+            break;
+        case LWS_CALLBACK_SERVER_WRITEABLE:
+            server->onWritable(wsi);
+            break;
+        default:
+            break;
+    }
+
+
+    return 0;
+}
+
+}
+
+namespace webfuse_test
+{
+
+class FakeAdapterServer::Private: public IServer
+{
+public:
+    explicit Private(int port)
+    : client_wsi(nullptr)
+    , message_received(false)
+    {
+        memset(ws_protocols, 0, sizeof(struct lws_protocols) * 2);
+        ws_protocols[0].name = "fs";
+        ws_protocols[0].callback = &wf_test_fake_adapter_server_callback;
+        ws_protocols[0].per_session_data_size = 0;
+        ws_protocols[0].user = reinterpret_cast<void*>(this);
+
+        memset(&info, 0, sizeof(struct lws_context_creation_info));
+        info.port = port;
+        info.mounts = NULL;
+        info.protocols =ws_protocols;
+        info.vhost_name = "localhost";
+        info.ws_ping_pong_interval = 10;
+        info.options = LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;
+
+	    context = lws_create_context(&info);
+    }
+
+    virtual ~Private()
+    {
+        lws_context_destroy(context);
+    }
+
+    void waitForConnection()
+    {
+        TimeoutWatcher watcher(DEFAULT_TIMEOUT);
+
+        while (nullptr == client_wsi)
+        {
+            watcher.check();
+            lws_service(context, 100);
+        }
+    }
+
+    void onConnectionEstablished(struct lws * wsi) override
+    {
+        client_wsi = wsi;
+    }
+
+    void onConnectionClosed(struct lws * wsi) override
+    {
+        if (wsi == client_wsi)
+        {
+            client_wsi = nullptr;
+        }
+    }
+
+    void onMessageReceived(struct lws * wsi, char const * data, size_t length) override
+    {
+        if (wsi == client_wsi)
+        {
+            last_message.assign(length, *data);
+            message_received = true;
+        }
+    }
+
+    void onWritable(struct lws * wsi) override
+    {
+        if (!queue.empty())
+        {
+            std::string const & message = queue.front();
+
+            unsigned char * data = new unsigned char[LWS_PRE + message.size()];
+            memcpy(&data[LWS_PRE], message.c_str(), message.size());
+            lws_write(wsi, &data[LWS_PRE], message.size(), LWS_WRITE_TEXT);
+            delete[] data;
+
+            queue.pop();
+            if (!queue.empty())
+            {
+                lws_callback_on_writable(wsi);
+            }
+        }
+    }
+
+
+private:
+    void send(std::string const & message)
+    {
+        if (nullptr != client_wsi)
+        {
+            queue.push(message);
+            lws_callback_on_writable(client_wsi);
+        }
+    }
+
+    struct lws * client_wsi;
+    bool message_received;
+
+    struct lws_protocols ws_protocols[2];
+	struct lws_context_creation_info info;
+    struct lws_context * context;
+    std::vector<char> last_message;
+    std::queue<std::string> queue;
+
+};
+
+FakeAdapterServer::FakeAdapterServer(int port)
+: d(new Private(port))
+{
+
+}
+
+FakeAdapterServer::~FakeAdapterServer()
+{
+    delete d;
+}
+
+void FakeAdapterServer::waitForConnection()
+{
+    d->waitForConnection();
+}
+
+}

test/fake_adapter_server.hpp

@@ -0,0 +1,24 @@
+#ifndef WF_TEST_FAKE_SERVER_HPP
+#define WF_TEST_FAKE_SERVER_HPP
+
+#include <libwebsockets.h>
+
+namespace webfuse_test
+{
+
+class FakeAdapterServer
+{
+    FakeAdapterServer(FakeAdapterServer const &) = delete;
+    FakeAdapterServer & operator=(FakeAdapterServer const &) = delete;
+public:
+    explicit FakeAdapterServer(int port);
+    ~FakeAdapterServer();
+    void waitForConnection();
+private:
+    class Private;
+    Private * d;
+};
+
+}
+
+#endif

test/provider/test_client_protocol.cc

@@ -0,0 +1,72 @@
+#include <gtest/gtest.h>
+#include <gmock/gmock.h>
+
+#include <webfuse/provider/client_protocol.h>
+#include <webfuse/provider/client_config.h>
+#include "fake_adapter_server.hpp"
+
+#include <cstring>
+#include <thread>
+#include <atomic>
+
+using webfuse_test::FakeAdapterServer;
+using testing::_;
+
+namespace
+{
+
+struct Context
+{
+    lws_context * context;
+    std::atomic<bool> isShutdownRequested;
+};
+
+void run(Context * context)
+{
+    while (!context->isShutdownRequested)
+    {
+        lws_service(context->context, 100);
+    }
+}
+
+
+}
+
+
+TEST(client_protocol, connect)
+{
+    FakeAdapterServer server(54321);
+
+    wfp_client_config * config = wfp_client_config_create();
+    wfp_client_protocol * protocol = wfp_client_protocol_create(config);
+
+    struct lws_protocols protocols[2];
+    memset(protocols, 0, sizeof(struct lws_protocols) * 2);
+    protocols[0].name = "fs";
+    wfp_client_protocol_init_lws(protocol, &protocols[0]);
+
+    struct lws_context_creation_info info;
+    memset(&info, 0, sizeof(struct lws_context_creation_info));
+    info.port = CONTEXT_PORT_NO_LISTEN;
+    info.protocols = protocols;
+    info.uid = -1;
+    info.gid = -1;
+
+    struct lws_context * context = lws_create_context(&info);
+    wfp_client_protocol_connect(protocol, context, "ws://localhost:54321/");
+
+    Context ctx;
+    ctx.context = context;
+    ctx.isShutdownRequested = false;
+    std::thread client_thread(run, &ctx);
+
+    server.waitForConnection();
+    
+    ctx.isShutdownRequested = true;
+    client_thread.join();
+
+    lws_context_destroy(context);
+
+    wfp_client_protocol_dispose(protocol);
+    wfp_client_config_dispose(config);
+}

test/timeout_watcher.cc

@@ -0,0 +1,44 @@
+#include "timeout_watcher.hpp"
+#include <stdexcept>
+
+using std::chrono::milliseconds;
+using std::chrono::duration_cast;
+using std::chrono::steady_clock;
+
+namespace
+{
+    milliseconds now()
+    {
+        return duration_cast<milliseconds>(steady_clock::now().time_since_epoch());
+    }
+}
+
+namespace webfuse_test
+{
+
+TimeoutWatcher::TimeoutWatcher(milliseconds timeout)
+: startedAt(now())
+, timeout_(timeout)
+{
+
+}
+
+TimeoutWatcher::~TimeoutWatcher()
+{
+
+}
+
+bool TimeoutWatcher::isTimeout()
+{
+    return (now() - startedAt) > timeout_;
+}
+
+void TimeoutWatcher::check()
+{
+    if (isTimeout())
+    {
+        throw std::runtime_error("timeout");
+    }
+}
+
+}

test/timeout_watcher.hpp

@@ -0,0 +1,26 @@
+#ifndef WF_TEST_TIMEOUT_WATCHER_HPP
+#define WF_TEST_TIMEOUT_WATCHER_HPP
+
+#include <chrono>
+
+namespace webfuse_test
+{
+
+class TimeoutWatcher final
+{
+    TimeoutWatcher(TimeoutWatcher const & other) = delete;
+    TimeoutWatcher& operator=(TimeoutWatcher const & other) = delete;
+public:
+    explicit TimeoutWatcher(std::chrono::milliseconds timeout);
+    ~TimeoutWatcher();
+    bool isTimeout();
+    void check();
+private:
+    std::chrono::milliseconds startedAt;
+    std::chrono::milliseconds timeout_;
+};
+
+}
+
+
+#endif