added implementation of wf_client_connect and wf_client_disconnect

lib/webfuse/adapter/impl/client.c

@@ -100,15 +100,14 @@ wf_impl_client_connect(
     struct wf_client * client,
     char const * url)
 {
-    (void) url;
-    wf_impl_client_protocol_callback(&client->protocol, WF_CLIENT_DISCONNECTED, NULL);
+    wf_impl_client_protocol_connect(&client->protocol, client->context, url);
 }
 
 void
 wf_impl_client_disconnect(
     struct wf_client * client)
 {
-    (void) client;
+    wf_impl_client_protocol_disconnect(&client->protocol);
 }
 
 void

lib/webfuse/adapter/impl/client_protocol.c

@@ -1,19 +1,56 @@
 #include "webfuse/adapter/impl/client_protocol.h"
 #include "webfuse/adapter/client_callback.h"
 #include "webfuse/core/protocol_names.h"
+#include "webfuse/core/url.h"
 #include "webfuse/core/util.h"
 
 #include <stddef.h>
 #include <libwebsockets.h>
 
 static int wf_impl_client_protocol_lws_callback(
-	struct lws * WF_UNUSED_PARAM(wsi),
-	enum lws_callback_reasons WF_UNUSED_PARAM(reason),
+	struct lws * wsi,
+	enum lws_callback_reasons reason,
 	void * WF_UNUSED_PARAM(user),
-	void * WF_UNUSED_PARAM(in),
+	void * in,
 	size_t WF_UNUSED_PARAM(len))
 {
-    return 0;
+    int result = 0;
+    struct lws_protocols const * ws_protocol = lws_get_protocol(wsi);
+    struct wf_client_protocol * protocol = (NULL != ws_protocol) ? ws_protocol->user : NULL;
+
+    if (NULL != protocol)
+    {
+        switch (reason)
+        {
+            case LWS_CALLBACK_CLIENT_ESTABLISHED:
+                protocol->is_connected = true;
+                protocol->callback(protocol->user_data, WF_CLIENT_CONNECTED, NULL);
+                break;
+            case LWS_CALLBACK_CLIENT_CONNECTION_ERROR:
+                protocol->is_connected = false;
+                protocol->callback(protocol->user_data, WF_CLIENT_DISCONNECTED, NULL);
+                break;
+            case LWS_CALLBACK_CLIENT_CLOSED:
+                protocol->is_connected = false;
+                protocol->callback(protocol->user_data, WF_CLIENT_DISCONNECTED, NULL);
+                protocol->wsi = NULL;
+                break;
+            case LWS_CALLBACK_SERVER_WRITEABLE:
+                // fall-through
+            case LWS_CALLBACK_CLIENT_WRITEABLE:
+                if (wsi == protocol->wsi)
+                {
+                    if (protocol->is_shutdown_requested)
+                    {
+                        result = 1;
+                    }
+                }
+            default:
+                break;
+        }
+    }
+
+    return result;
 }
 
 void
@@ -22,6 +59,9 @@ wf_impl_client_protocol_init(
     wf_client_callback_fn * callback,
     void * user_data)
 {
+    protocol->is_connected = false,
+    protocol->is_shutdown_requested = false;
+    protocol->wsi = NULL;
     protocol->callback = callback;
     protocol->user_data = user_data;
     protocol->callback(protocol->user_data, WF_CLIENT_INIT, NULL);
@@ -57,14 +97,47 @@ wf_impl_client_protocol_init_lws(
 void
 wf_impl_client_protocol_connect(
     struct wf_client_protocol * protocol,
+    struct lws_context * context,
     char const * url)
 {
+    struct wf_url url_data;
+    bool const success = wf_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 = WF_PROTOCOL_NAME_PROVIDER_SERVER;
+        info.local_protocol_name = WF_PROTOCOL_NAME_ADAPTER_CLIENT;
+        info.pwsi = &protocol->wsi;
 
+        lws_client_connect_via_info(&info);
+        wf_url_cleanup(&url_data);
+    }
+    else
+    {
+        protocol->callback(protocol->user_data, WF_CLIENT_DISCONNECTED, NULL);
+    }
 }
 
 void
 wf_impl_client_protocol_disconnect(
     struct wf_client_protocol * protocol)
 {
+    if (protocol->is_connected)
+    {
+        protocol->is_shutdown_requested = true;
+        lws_callback_on_writable(protocol->wsi);
+    }
+    else
+    {
+        protocol->callback(protocol->user_data, WF_CLIENT_DISCONNECTED, NULL);
+    }
     
 }

lib/webfuse/adapter/impl/client_protocol.h

@@ -3,12 +3,17 @@
 
 #include "webfuse/adapter/client_callback.h"
 
+#ifndef __cplusplus
+#include <stdbool.h>
+#endif
+
 #ifdef __cplusplus
 extern "C"
 {
 #endif
 
 struct lws_protocols;
+struct lws_context;
 
 typedef void
 wf_client_protocol_callback_fn(
@@ -18,6 +23,9 @@ wf_client_protocol_callback_fn(
 
 struct wf_client_protocol
 {
+    bool is_connected;
+    bool is_shutdown_requested;
+    struct lws * wsi;
     wf_client_callback_fn * callback;
     void * user_data;
 };
@@ -46,6 +54,7 @@ wf_impl_client_protocol_init_lws(
 extern void
 wf_impl_client_protocol_connect(
     struct wf_client_protocol * protocol,
+    struct lws_context * conext,
     char const * url);
 
 extern void

lib/webfuse/adapter/impl/client_tlsconfig.c

@@ -1,5 +1,4 @@
 #include "webfuse/adapter/impl/client_tlsconfig.h"
-#include "webfuse/core/url.h"
 
 #include <stdlib.h>
 #include <string.h>

meson.build

@@ -210,6 +210,7 @@ alltests = executable('alltests',
 	'test/webfuse/utils/path.c',
 	'test/webfuse/utils/static_filesystem.c',
 	'test/webfuse/utils/ws_server.cc',
+	'test/webfuse/utils/threaded_ws_server.cc',
 	'test/webfuse/mocks/fake_invokation_context.cc',
 	'test/webfuse/mocks/mock_authenticator.cc',
 	'test/webfuse/mocks/mock_request.cc',

test/webfuse/tests/adapter/test_client.cc

@@ -2,6 +2,9 @@
 
 #include "webfuse/adapter/client.h"
 #include "webfuse/adapter/credentials.h"
+#include "webfuse/utils/threaded_ws_server.h"
+
+using webfuse_test::ThreadedWsServer;
 
 namespace
 {
@@ -63,18 +66,42 @@ void callback(
     }
 }
 
+void callback2(
+    wf_client * client,
+    int reason,
+    void * args)
+{
+    auto * ctx = reinterpret_cast<context*>(wf_client_get_userdata(client));
+
+    switch (reason)
+    {
+        case WF_CLIENT_CREATED:
+            ctx->state = connection_state::connecting;
+            break;
+        case WF_CLIENT_CONNECTED:
+            ctx->state = connection_state::connected;
+            break;
+        case WF_CLIENT_DISCONNECTED:
+            ctx->state = connection_state::disconnected;
+            break;
+        default:
+            break;
+    }
+}
+
+
 }
 
 TEST(client, general_usage)
 {
+    ThreadedWsServer server(54321);
+
     context ctx;
     ctx.state = connection_state::connecting;
 
     wf_client * client = wf_client_create(
         &callback, reinterpret_cast<void*>(&ctx));
 
-    if (nullptr != client)
-    {
     while (ctx.state != connection_state::disconnected)
     {
         wf_client_service(client);
@@ -82,4 +109,28 @@ TEST(client, general_usage)
 
     wf_client_dispose(client);
 }
+
+TEST(client, connect)
+{
+    ThreadedWsServer server(54321);
+
+    context ctx;
+    ctx.state = connection_state::connecting;
+
+    wf_client * client = wf_client_create(
+        &callback2, reinterpret_cast<void*>(&ctx));
+
+    wf_client_connect(client, "ws://localhost:54321/");
+    while (ctx.state != connection_state::connected)
+    {
+        wf_client_service(client);
+    }
+
+    wf_client_disconnect(client);
+    while (ctx.state != connection_state::disconnected)
+    {
+        wf_client_service(client);
+    }
+
+    wf_client_dispose(client);
 }

test/webfuse/utils/threaded_ws_server.cc

@@ -0,0 +1,197 @@
+#include "webfuse/utils/threaded_ws_server.h"
+#include "webfuse/core/protocol_names.h"
+#include "webfuse/core/lws_log.h"
+
+#include <libwebsockets.h>
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <chrono>
+
+
+#define TIMEOUT (std::chrono::milliseconds(10 * 1000))
+
+namespace
+{
+
+class IServer
+{
+public:
+    virtual ~IServer() = default;
+    virtual void OnConnected(lws * wsi) = 0;
+    virtual void OnConnectionClosed(lws * wsi) = 0;
+};
+
+}
+
+namespace webfuse_test
+{
+
+class ThreadedWsServer::Private : IServer
+{
+    Private(Private const &) = delete;
+    Private & operator=(Private const &) = delete;
+public:
+    explicit Private(int port);
+    ~Private();
+    void WaitForConnection();
+    void OnConnected(lws * wsi) override;
+    void OnConnectionClosed(lws * wsi) override;
+
+private:
+    static void run(Private * self);
+    int port_;
+    bool is_connected;
+    bool is_shutdown_requested;
+    lws * wsi_;
+    lws_context * ws_context;
+    lws_protocols ws_protocols[2];
+	lws_context_creation_info info;
+    std::thread context;
+    std::mutex mutex;
+    std::condition_variable convar;
+};
+
+}
+
+extern "C"
+{
+
+static int wf_test_utils_threaded_ws_server_callback(
+    struct lws * wsi,
+    enum lws_callback_reasons reason,
+    void * user,
+    void * in,
+    size_t len)
+{
+    int result = 0;
+    struct lws_protocols const * ws_protocol = lws_get_protocol(wsi);
+    auto * server = reinterpret_cast<IServer*>(nullptr != ws_protocol ? ws_protocol->user : nullptr);
+
+    if (nullptr != server)
+    {
+        switch (reason)
+        {
+            case LWS_CALLBACK_ESTABLISHED:
+                server->OnConnected(wsi);
+                break;
+            case LWS_CALLBACK_CLOSED:
+                server->OnConnectionClosed(wsi);
+                break;
+            default:
+                break;
+        }
+    }
+
+    return result;
+}
+
+}
+
+
+namespace webfuse_test
+{
+
+ThreadedWsServer::ThreadedWsServer(int port)
+: d(new Private(port))
+{
+
+}
+
+ThreadedWsServer::~ThreadedWsServer()
+{
+    delete d;
+}
+
+void ThreadedWsServer::WaitForConnection()
+{
+    d->WaitForConnection();
+}
+
+ThreadedWsServer::Private::Private(int port)
+: port_(port)
+, is_connected(false)
+, is_shutdown_requested(false)
+, wsi_(nullptr)
+{
+    wf_lwslog_disable();
+    IServer * server = this;
+    memset(ws_protocols, 0, sizeof(struct lws_protocols) * 2 );
+
+    ws_protocols[0].name = WF_PROTOCOL_NAME_PROVIDER_SERVER;
+    ws_protocols[0].callback = &wf_test_utils_threaded_ws_server_callback;
+    ws_protocols[0].per_session_data_size = 0;
+    ws_protocols[0].user = reinterpret_cast<void*>(server);
+
+    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;
+
+    ws_context = lws_create_context(&info);
+
+    context = std::thread(&run, this);
+}
+
+ThreadedWsServer::Private::~Private()
+{
+    {
+        std::unique_lock<std::mutex> lock(mutex);
+        is_shutdown_requested = true;
+    }
+
+    lws_cancel_service(ws_context);
+    context.join();
+    lws_context_destroy(ws_context);
+}
+
+void ThreadedWsServer::Private::run(Private * self)
+{
+    bool is_running = true;
+    while (is_running)
+    {
+        lws_service(self->ws_context, 0);
+        {
+            std::unique_lock<std::mutex> lock(self->mutex);
+            is_running = !self->is_shutdown_requested;
+        }
+    }
+}
+
+void ThreadedWsServer::Private::WaitForConnection()
+{
+    std::unique_lock<std::mutex> lock(mutex);
+    while (!is_connected)
+    {
+        auto status = convar.wait_for(lock, TIMEOUT);
+        if (std::cv_status::timeout == status)
+        {
+            throw std::runtime_error("timeout");
+        }
+    }
+}
+
+void ThreadedWsServer::Private::OnConnected(lws * wsi)
+{
+    std::unique_lock<std::mutex> lock(mutex);
+    is_connected = true;
+    wsi_ = wsi;
+    convar.notify_all();
+}
+
+void ThreadedWsServer::Private::OnConnectionClosed(lws * wsi)
+{
+    std::unique_lock<std::mutex> lock(mutex);
+    if (wsi == wsi_)
+    {
+        is_connected = false;
+        wsi_ = wsi;
+        convar.notify_all();
+    }
+}
+
+
+}

test/webfuse/utils/threaded_ws_server.h

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