made it single threaded (still buggy)

CMakeLists.txt

@@ -41,17 +41,22 @@ set(EXTRA_CFLAGS
 add_library(fuse-wsfs
 	src/wsfs/status.c
 	src/wsfs/filesystem.c
+	src/wsfs/server.c
+	src/wsfs/message.c
+	src/wsfs/message_queue.c
 	src/wsfs/operation/lookup.c
 	src/wsfs/operation/getattr.c
 	src/wsfs/operation/readdir.c
 	src/wsfs/operation/open.c
 	src/wsfs/operation/close.c
 	src/wsfs/operation/read.c
-	src/wsfs/response_parser.c
-	src/wsfs/server.c
 	src/wsfs/server_config.c
 	src/wsfs/server_protocol.c
-	src/wsfs/jsonrpc.c
+	src/wsfs/jsonrpc/server.c
+	src/wsfs/jsonrpc/method.c
+	src/wsfs/jsonrpc/request.c
+	src/wsfs/jsonrpc/response.c
+	src/wsfs/jsonrpc/util.c
 )
 
 target_include_directories(fuse-wsfs PUBLIC src ${EXTRA_INCLUDE_DIRS})

src/wsfs/filesystem.c

@@ -1,6 +1,6 @@
 #include "wsfs/filesystem.h"
 #include "wsfs/operations.h"
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
 
 #include <stddef.h>
 #include <string.h>
@@ -17,16 +17,19 @@ static struct fuse_lowlevel_ops const wsfs_filesystem_operations =
 };
 
 
-void wsfs_filesystem_init(
+bool wsfs_filesystem_init(
     struct wsfs_filesystem * filesystem,
+	struct wsfs_jsonrpc_server * rpc,
     char * mount_point)
 {
+	bool result = false;
+
 	char * argv[] = {"", NULL};
 	filesystem->args.argc = 1;
 	filesystem->args.argv = argv;
 	filesystem->args.allocated = 0;
 
-	filesystem->user_data.rpc = wsfs_jsonrpc_create(NULL, NULL, NULL);
+	filesystem->user_data.rpc = rpc;
 	filesystem->user_data.timeout = 1.0;
 	memset(&filesystem->buffer, 0, sizeof(struct fuse_buf));
 
@@ -37,9 +40,10 @@ void wsfs_filesystem_init(
         &filesystem->user_data);
 	if (NULL != filesystem->session)
 	{
-		fuse_session_mount(filesystem->session, mount_point);
+		result = (0 == fuse_session_mount(filesystem->session, mount_point));
 	}
 
+	return result;
 }
 
 void wsfs_filesystem_cleanup(

src/wsfs/filesystem.h

@@ -1,9 +1,15 @@
 #ifndef _WSFS_FILESYSTEM_H
 #define _WSFS_FILESYSTEM_H
 
+#ifndef __cplusplus
+#include <stdbool.h>
+#endif
+
 #include "wsfs/fuse_wrapper.h"
 #include "wsfs/operations.h"
 
+struct wsfs_jsonrpc_server;
+
 struct wsfs_filesystem
 {
 	struct fuse_args args;
@@ -12,8 +18,14 @@ struct wsfs_filesystem
 	struct wsfs_operations_context user_data;
 };
 
-extern void wsfs_filesystem_init(
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern bool wsfs_filesystem_init(
     struct wsfs_filesystem * filesystem,
+    struct wsfs_jsonrpc_server * rpc,
     char * mount_point);
 
 extern void wsfs_filesystem_cleanup(
@@ -25,4 +37,9 @@ extern int wsfs_filesystem_get_fd(
 extern void wsfs_filesystem_process_request(
     struct wsfs_filesystem * filesystem);
 
+#ifdef __cplusplus
+}
+#endif
+
+
 #endif

src/wsfs/jsonrpc.c

@@ -1,246 +0,0 @@
-#include "wsfs/jsonrpc.h"
-
-#include <stdio.h>
-#include <stdlib.h>
-#include <time.h>
-
-#include <pthread.h>
-
-#include "wsfs/response_parser.h"
-
-#define DEFAULT_TIMEOUT_SECS 10
-
-struct wsfs_jsonrpc_response
-{
-	int id;
-	wsfs_status status;
-	json_t * result;
-};
-
-struct wsfs_jsonrpc
-{
-	pthread_mutex_t lock;
-	pthread_cond_t finished;
-	pthread_condattr_t finished_attributes;
-
-	wsfs_create_message_fn * create_message;
-	wsfs_send_message_fn * send_message;
-	void * user_data;
-
-	bool is_finished;
-	struct wsfs_jsonrpc_response response;	
-};
-
-static json_t * wsfs_request_create(
-	char const * method,
-	int id,
-	char const * param_info,
-	va_list args)
-{
-	json_t * request = json_object();
-	json_object_set_new(request, "method", json_string(method));
-	json_t * params = json_array();
-	
-	for (char const * param_type = param_info; '\0' != *param_type; param_type++)
-	{
-		switch(*param_type)
-		{
-			case 's':
-			{
-				char const * const value = va_arg(args, char const *);
-				json_array_append_new(params, json_string(value));
-			}
-			break;
-			case 'i':
-			{
-				int const value = va_arg(args, int);
-				json_array_append_new(params, json_integer(value));
-			}
-			break;
-			default:
-			fprintf(stderr, "fatal: unknown param_type '%c'\n", *param_type);
-			exit(EXIT_FAILURE);
-			break;
-		}
-	}
-	
-
-	json_object_set_new(request, "params", params);
-	json_object_set_new(request, "id", json_integer(id));
-
-	return request;
-}
-
-
-wsfs_status wsfs_jsonrpc_invoke(
-	struct wsfs_jsonrpc * rpc,
-	json_t * * result,
-	char const * method,
-	char const * param_info,
-	...
-)
-{
-	return WSFS_BAD_NOTIMPLEMENTED;
-
-	// enqueue message
-	pthread_mutex_lock(&rpc->lock);
-	wsfs_status status = WSFS_BAD;
-
-	char * message = NULL;
-	size_t length = 0;
-	if (-1 == rpc->response.id) {
-
-		va_list args;
-		va_start(args, param_info);
-		json_t * request = wsfs_request_create(method, 42, param_info, args);	
-		va_end(args);
-
-		
-		length = json_dumpb(request, NULL, 0, JSON_COMPACT);
-		if (0 < length)
-		{	
-			rpc->is_finished = false;	
-			rpc->response.id = 42;
-			rpc->response.result = NULL;
-			rpc->response.status = WSFS_GOOD;
-
-			message = rpc->create_message(length);
-			json_dumpb(request, message, length, JSON_COMPACT);
-
-		}
-		json_decref(request);
-	}
-	pthread_mutex_unlock(&rpc->lock);
-	
-
-	if (NULL != message)
-	{
-		bool const success = rpc->send_message(message, length, rpc->user_data);
-
-		// wait for answer
-		pthread_mutex_lock(&rpc->lock);
-
-		if (success)
-		{
-			struct timespec timeout;
-			clock_gettime(CLOCK_MONOTONIC, &timeout);
-			timeout.tv_sec += DEFAULT_TIMEOUT_SECS;
-			int rc = 0;
-			while ((0 == rc) && (!rpc->is_finished)) {
-				rc = pthread_cond_timedwait(&rpc->finished, &rpc->lock, &timeout);
-			}
-
-			if (rpc->is_finished)
-			{
-				status = rpc->response.status;
-				*result = rpc->response.result;
-			}
-			else
-			{
-				status = WSFS_BAD_TIMEOUT;
-			}
-		}
-		
-		rpc->response.id = -1;
-		rpc->response.result = NULL;
-		rpc->response.status = WSFS_GOOD;
-		pthread_mutex_unlock(&rpc->lock);
-	}
-
-	return status;
-}
-
-struct wsfs_jsonrpc * wsfs_jsonrpc_create(
-	wsfs_create_message_fn * create_message,
-	wsfs_send_message_fn * send_message,
-	void * user_data)
-{
-	struct wsfs_jsonrpc * rpc = malloc(sizeof(struct wsfs_jsonrpc));
-	if (NULL != rpc)
-	{
-		pthread_mutex_init(&rpc->lock, NULL);
-
-		pthread_condattr_init(&rpc->finished_attributes);
-		pthread_condattr_setclock(&rpc->finished_attributes, CLOCK_MONOTONIC);
-		pthread_cond_init(&rpc->finished, &rpc->finished_attributes);
-
-		rpc->create_message = create_message;
-		rpc->send_message = send_message;
-		rpc->user_data = user_data;
-		rpc->is_finished = true;
-		rpc->response.id = -1;
-		rpc->response.status = WSFS_GOOD;
-		rpc->response.result = NULL;
-	}
-
-	return rpc;
-}
-
-void wsfs_jsonrpc_set_user_data(
-	struct wsfs_jsonrpc * rpc,
-	void * user_data)
-{
-	rpc->user_data = user_data;
-}
-
-void wsfs_jsonrpc_dispose(
-	struct wsfs_jsonrpc * rpc)
-{
-	if (NULL != rpc->response.result)
-	{
-		json_decref(rpc->response.result);
-	}
-
-	pthread_cond_destroy(&rpc->finished);	
-	pthread_condattr_destroy(&rpc->finished_attributes);
-	pthread_mutex_destroy(&rpc->lock);
-	
-
-	free(rpc);
-}
-
-void wsfs_jsonrpc_on_message(
-	char const * message,
-	size_t length,
-	void * user_data)
-{
-	struct wsfs_jsonrpc * rpc = user_data;
-
-	struct wsfs_response response;
-	wsfs_response_parse(message, length, &response);
-
-	if (-1 != response.id)
-	{
-		pthread_mutex_lock(&rpc->lock);
-
-		if (response.id == rpc->response.id)
-		{
-			rpc->is_finished = true;
-			rpc->response.status = response.status;
-			rpc->response.result = response.result;
-		}
-		else
-		{
-			if (NULL != response.result)
-			{
-				json_decref(response.result);
-			}
-		}
-
-		pthread_cond_signal(&rpc->finished);
-		pthread_mutex_unlock(&rpc->lock);
-	}
-}
-
-int wsfs_json_get_int(json_t * object, char const * key, int default_value)
-{
-	int result = default_value;
-
-	json_t * holder = json_object_get(object, key);
-	if ((NULL != holder) && (json_is_integer(holder)))
-	{
-		result = json_integer_value(holder);
-	}
-
-	return result;
-}

src/wsfs/jsonrpc.h

@@ -1,59 +0,0 @@
-#ifndef _WSFS_JSONRPC_H
-#define _WSFS_JSONRPC_H
-
-#ifndef __cplusplus
-#include <stdarg.h>
-#include <stddef.h>
-#include <stdbool.h>
-#else
-#include <cstdarg>
-#include <cstddef>
-using std::size_t;
-#endif
-
-#include <jansson.h>
-#include "wsfs/status.h"
-
-struct wsfs_jsonrpc;
-
-typedef char * wsfs_create_message_fn(size_t size);
-typedef bool wsfs_send_message_fn(char * message, size_t length, void * user_data);
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-extern struct wsfs_jsonrpc * wsfs_jsonrpc_create(
-	wsfs_create_message_fn * create_message,
-	wsfs_send_message_fn * send_message,
-	void * user_data);
-
-extern void wsfs_jsonrpc_set_user_data(
-	struct wsfs_jsonrpc * rpc,
-	void * user_data);
-
-extern void wsfs_jsonrpc_dispose(
-	struct wsfs_jsonrpc * rpc);
-
-extern wsfs_status wsfs_jsonrpc_invoke(
-	struct wsfs_jsonrpc * rpc,
-	json_t * * result,
-	char const * method,
-	char const * param_info,
-	...
-);
-
-extern void wsfs_jsonrpc_on_message(
-	char const * message,
-	size_t length,
-	void * user_data);
-
-
-extern int wsfs_json_get_int(json_t * object, char const * key, int default_value);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif
-

src/wsfs/jsonrpc/method.c

@@ -0,0 +1,28 @@
+#include "wsfs/jsonrpc/method_intern.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+struct wsfs_jsonrpc_method * wsfs_jsonrpc_method_create(
+    char const * name,
+    wsfs_jsonrpc_method_invoke_fn * invoke,
+    void * user_data)
+{
+    struct wsfs_jsonrpc_method * method = malloc(sizeof(struct wsfs_jsonrpc_method));
+    if (NULL != method)
+    {
+        method->next = NULL;
+        method->name = strdup(name);
+        method->invoke = invoke;
+        method->user_data = user_data;
+    }
+
+    return method;
+}
+
+void wsfs_jsonrpc_method_dispose(
+    struct wsfs_jsonrpc_method * method)
+{
+    free(method->name);
+    free(method);
+}

src/wsfs/jsonrpc/method.h

@@ -0,0 +1,24 @@
+#ifndef _WSFS_JSONRPC_METHOD_H
+#define _WSFS_JSONRPC_METHOD_H
+
+#ifndef __cplusplus
+#include <stdbool.h>
+#endif
+
+#include <jansson.h>
+#include "wsfs/status.h"
+
+
+typedef bool wsfs_jsonrpc_method_invoke_fn(
+    void * user_data,
+    struct json_t const * method_call);
+
+typedef void wsfs_jsonrpc_method_finished_fn(
+	void * user_data,
+	wsfs_status status,
+	struct json_t const * result);
+
+
+
+
+#endif

src/wsfs/jsonrpc/method_intern.h

@@ -0,0 +1,31 @@
+#ifndef _WSFS_JSONRPC_METHOD_INTERN_H
+#define _WSFS_JSONRPC_METHOD_INTERN_H
+
+#include "wsfs/jsonrpc/method.h"
+
+struct wsfs_jsonrpc_method
+{
+    struct wsfs_jsonrpc_method * next;
+    char * name;
+    wsfs_jsonrpc_method_invoke_fn * invoke;
+    void * user_data;
+};
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern struct wsfs_jsonrpc_method * wsfs_jsonrpc_method_create(
+    char const * name,
+    wsfs_jsonrpc_method_invoke_fn * invoke,
+    void * user_data);
+
+extern void wsfs_jsonrpc_method_dispose(
+    struct wsfs_jsonrpc_method * method);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/wsfs/jsonrpc/request.c

@@ -0,0 +1,41 @@
+#include "wsfs/jsonrpc/request.h"
+
+json_t * wsfs_jsonrpc_request_create(
+	char const * method,
+	int id,
+	char const * param_info,
+	va_list args)
+{
+	json_t * request = json_object();
+	json_object_set_new(request, "method", json_string(method));
+	json_t * params = json_array();
+	
+	for (char const * param_type = param_info; '\0' != *param_type; param_type++)
+	{
+		switch(*param_type)
+		{
+			case 's':
+			{
+				char const * const value = va_arg(args, char const *);
+				json_array_append_new(params, json_string(value));
+			}
+			break;
+			case 'i':
+			{
+				int const value = va_arg(args, int);
+				json_array_append_new(params, json_integer(value));
+			}
+			break;
+			default:
+			fprintf(stderr, "fatal: unknown param_type '%c'\n", *param_type);
+			exit(EXIT_FAILURE);
+			break;
+		}
+	}
+	
+
+	json_object_set_new(request, "params", params);
+	json_object_set_new(request, "id", json_integer(id));
+
+	return request;
+}

src/wsfs/jsonrpc/request.h

@@ -0,0 +1,31 @@
+#ifndef _WSFS_JSONRPC_REQUEST_H
+#define _WSFS_JSONRPC_REQUEST_H
+
+#ifndef __cplusplus
+#include <stdarg.h>
+#include <stddef.h>
+#include <stdbool.h>
+#else
+#include <cstdarg>
+#include <cstddef>
+using std::size_t;
+#endif
+
+#include <jansson.h>
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern json_t * wsfs_jsonrpc_request_create(
+	char const * method,
+	int id,
+	char const * param_info,
+	va_list args);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/wsfs/jsonrpc/response.c

@@ -1,9 +1,9 @@
-#include "wsfs/response_parser.h"
+#include "wsfs/jsonrpc/response.h"
 
-void wsfs_response_parse(
+void wsfs_jsonrpc_response_init(
+	struct wsfs_jsonrpc_response * result,
     char const * buffer,
-	size_t length,
+	size_t length)
-	struct wsfs_response * result)
 {
 	result->status = WSFS_BAD;
 	result->id = -1;
@@ -49,3 +49,11 @@ void wsfs_response_parse(
 	json_decref(response);
 }
 
+void wsfs_jsonrpc_response_cleanup(
+	struct wsfs_jsonrpc_response * response)
+{
+    if (NULL != response->result)
+    {
+        json_decref(response->result);
+    }
+}

src/wsfs/jsonrpc/response.h

@@ -0,0 +1,38 @@
+#ifndef _WSFS_JSONRPC_RESPONSE_H
+#define _WFSF_JSONRPC_RESPONSE_H
+
+#ifndef __cplusplus
+#include <stddef.h>
+#else
+#include <cstddef>
+using std::size_t;
+#endif
+
+#include <jansson.h>
+#include "wsfs/status.h"
+
+struct wsfs_jsonrpc_response
+{
+	wsfs_status status;
+	int id;
+	json_t * result;	
+};
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern void wsfs_jsonrpc_response_init(
+	struct wsfs_jsonrpc_response * response,
+    char const * buffer,
+	size_t buffer_length);
+
+extern void wsfs_jsonrpc_response_cleanup(
+	struct wsfs_jsonrpc_response * response);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
+

src/wsfs/jsonrpc/server.c

@@ -0,0 +1,129 @@
+#include "wsfs/jsonrpc/server.h"
+#include <string.h>
+
+#include "wsfs/jsonrpc/method_intern.h"
+#include "wsfs/jsonrpc/request.h"
+#include "wsfs/jsonrpc/response.h"
+
+static struct wsfs_jsonrpc_method const * wsfs_jsonrpc_server_getmethod(
+    struct wsfs_jsonrpc_server * server,
+    char const * name)
+{
+    struct wsfs_jsonrpc_method * method = server->methods;
+    while ((NULL != method) && (0 == strcmp(name, method->name)))
+    {
+        method = method->next;
+    }
+
+    return method;
+}
+
+void wsfs_jsonrpc_server_init(
+    struct wsfs_jsonrpc_server * server)
+{
+    server->methods = NULL;
+    server->request.is_pending = false;
+}
+
+void wsfs_jsonrpc_server_cleanup(
+    struct wsfs_jsonrpc_server * server)
+{
+    if (server->request.is_pending)
+    {
+        server->request.finished(server->request.user_data, WSFS_BAD, NULL);
+        server->request.is_pending = false;
+    }
+
+    struct wsfs_jsonrpc_method * method = server->methods;
+    while (NULL != method)
+    {
+        struct wsfs_jsonrpc_method * next = method->next;
+        method->next = NULL;
+        wsfs_jsonrpc_method_dispose(method);
+        method = next;
+    }
+    server->methods = NULL;
+}
+
+void wsfs_jsonrpc_server_add(
+    struct wsfs_jsonrpc_server * server,
+    char const * name,
+    wsfs_jsonrpc_method_invoke_fn * invoke,
+    void * user_data)
+{
+    struct wsfs_jsonrpc_method * method = wsfs_jsonrpc_method_create(name, invoke, user_data);
+    method->next = server->methods;
+    server->methods = method;
+}
+
+void wsfs_jsonrpc_server_invoke(
+	struct wsfs_jsonrpc_server * server,
+	wsfs_jsonrpc_method_finished_fn * finished,
+	void * user_data,
+	char const * method_name,
+	char const * param_info,
+	...
+)
+{
+    if (!server->request.is_pending)
+    {
+        struct wsfs_jsonrpc_method const * method = wsfs_jsonrpc_server_getmethod(server, method_name);
+        if (NULL != method)
+        {
+            server->request.is_pending = true;
+            server->request.finished = finished;
+            server->request.user_data = user_data;
+            server->request.id = 42;
+            
+            va_list args;
+            va_start(args, param_info);
+            json_t * request = wsfs_jsonrpc_request_create(method_name, server->request.id, param_info, args);
+            va_end(args);
+            if (NULL != request)
+            {
+                if (!method->invoke(method->user_data, request))
+                {
+                    finished(user_data, WSFS_BAD, NULL);
+                }
+                json_decref(request);
+            }
+        }
+        else
+        {
+            finished(user_data, WSFS_BAD_NOTIMPLEMENTED, NULL);
+        }        
+    }
+    else
+    {
+        finished(user_data, WSFS_BAD_BUSY, NULL);
+    }
+    
+}
+
+void wsfs_jsonrpc_server_onresult(
+    struct wsfs_jsonrpc_server * server,
+    char const * message,
+    size_t length)
+{
+	struct wsfs_jsonrpc_response response;
+	wsfs_jsonrpc_response_init(&response, message, length);
+
+	if (-1 != response.id)
+	{
+		if ((server->request.is_pending) && (response.id == server->request.id))
+		{
+            wsfs_jsonrpc_method_finished_fn * finished = server->request.finished;
+            void * user_data = server->request.user_data;
+
+			server->request.is_pending = false;
+            server->request.id = 0;
+            server->request.user_data = NULL;
+            server->request.finished = NULL;
+
+            finished(user_data, response.status, response.result);
+		}
+    }
+
+    wsfs_jsonrpc_response_cleanup(&response);
+}
+

src/wsfs/jsonrpc/server.h

@@ -0,0 +1,68 @@
+#ifndef _WSFS_JSONRPC_SERVER_H
+#define _WSFS_JSONRPC_SERVER_H
+
+#ifndef __cplusplus
+#include <stdarg.h>
+#include <stddef.h>
+#include <stdbool.h>
+#else
+#include <cstdarg>
+#include <cstddef>
+using std::size_t;
+#endif
+
+#include <jansson.h>
+#include "wsfs/jsonrpc/method.h"
+
+struct wsfs_jsonrpc_request
+{
+    bool is_pending;
+    wsfs_jsonrpc_method_finished_fn * finished;
+    void * user_data;
+    int id;
+};
+
+struct wsfs_jsonrpc_server
+{
+    struct wsfs_jsonrpc_method * methods;
+    struct wsfs_jsonrpc_request request;
+};
+
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern void wsfs_jsonrpc_server_init(
+    struct wsfs_jsonrpc_server * server);
+
+extern void wsfs_jsonrpc_server_cleanup(
+    struct wsfs_jsonrpc_server * server);
+
+extern void wsfs_jsonrpc_server_add(
+    struct wsfs_jsonrpc_server * server,
+    char const * name,
+    wsfs_jsonrpc_method_invoke_fn * invoke,
+    void * user_data );
+
+extern void wsfs_jsonrpc_server_invoke(
+	struct wsfs_jsonrpc_server * server,
+	wsfs_jsonrpc_method_finished_fn * finished,
+	void * user_data,
+	char const * method,
+	char const * param_info,
+	...
+);
+
+extern void wsfs_jsonrpc_server_onresult(
+    struct wsfs_jsonrpc_server * server,
+    char const * message,
+    size_t length);
+
+#ifdef __cplusplus
+}
+#endif
+
+
+#endif

src/wsfs/jsonrpc/util.c

@@ -0,0 +1,14 @@
+#include "wsfs/jsonrpc/util.h"
+
+int wsfs_json_get_int(json_t const * object, char const * key, int default_value)
+{
+	int result = default_value;
+
+	json_t * holder = json_object_get(object, key);
+	if ((NULL != holder) && (json_is_integer(holder)))
+	{
+		result = json_integer_value(holder);
+	}
+
+	return result;
+}

src/wsfs/jsonrpc/util.h

@@ -0,0 +1,17 @@
+#ifndef _WSFS_JSON_UTIL_H
+#define _WFSF_JSON_UTIL_H
+
+#include <jansson.h>
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern int wsfs_json_get_int(json_t const * object, char const * key, int default_value);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/wsfs/message.c

@@ -0,0 +1,24 @@
+#include "wsfs/message.h"
+
+#include <stdlib.h>
+#include <libwebsockets.h>
+
+struct wsfs_message * wsfs_message_create(size_t length)
+{
+    char * data = malloc(sizeof(struct wsfs_message) + LWS_PRE + length);
+    struct wsfs_message * message = (struct wsfs_message *) data;
+    if (NULL != message)
+    {
+        message->data = &data[sizeof(struct wsfs_message) + LWS_PRE];
+        message->length = length;
+        message->next = NULL;
+    }
+
+    return message;
+}
+
+void wsfs_message_dispose(
+    struct wsfs_message * message)
+{
+    free(message);    
+}

src/wsfs/message.h

@@ -0,0 +1,32 @@
+#ifndef _WSFS_MESSAGE_H
+#define _WSFS_MESSAGE_H
+
+#ifndef __cplusplus
+#include <stddef.h>
+#else
+#include <cstddef>
+using std::size_t;
+#endif
+
+struct wsfs_message
+{
+    struct wsfs_message * next;
+    char * data;
+    size_t length;
+};
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern struct wsfs_message * wsfs_message_create(size_t length);
+
+extern void wsfs_message_dispose(
+    struct wsfs_message * message);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/wsfs/message_queue.c

@@ -0,0 +1,63 @@
+#include "wsfs/message_queue.h"
+#include "wsfs/message.h"
+
+void wsfs_message_queue_init(
+    struct wsfs_message_queue * queue)
+{
+    queue->first = NULL;
+    queue->last = NULL;
+    
+}
+
+void wsfs_message_queue_cleanup(
+    struct wsfs_message_queue * queue)
+{
+    struct wsfs_message * message = queue->first;
+    while (NULL != message)
+    {
+        struct wsfs_message * next = message->next;
+        wsfs_message_dispose(message);
+        message = next;
+    }
+    wsfs_message_queue_init(queue);
+}
+
+bool wsfs_message_queue_empty(
+    struct wsfs_message_queue * queue)
+{
+    return (NULL == queue->first);
+}
+
+void wsfs_message_queue_push(
+    struct wsfs_message_queue * queue,
+    struct wsfs_message * message)
+{
+    message->next = NULL;
+
+    if (NULL != queue->last)
+    {
+        queue->last->next = message;
+        queue->last = message;
+    }
+    else
+    {
+        queue->first = message;
+        queue->last = message;
+    }
+}
+
+struct wsfs_message * wsfs_message_queue_pop(
+    struct wsfs_message_queue * queue)
+{
+    struct wsfs_message * const result = queue->first;
+    if (NULL != result)
+    {
+        queue->first = queue->first->next;
+        if (NULL == queue->first)
+        {
+            queue->last = NULL;
+        }
+    } 
+
+    return result;
+}

src/wsfs/message_queue.h

@@ -0,0 +1,42 @@
+#ifndef _WSFS_MESSAGE_QUEUE_H
+#define _WSFS_MESSAGE_QUEUE_H
+
+#ifndef __cplusplus
+#include <stdbool.h>
+#endif
+
+struct wsfs_message_queue;
+struct wsfs_message;
+
+struct wsfs_message_queue
+{
+    struct wsfs_message * first;
+    struct wsfs_message * last;
+};
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+extern void wsfs_message_queue_init(
+    struct wsfs_message_queue * queue);
+
+extern void wsfs_message_queue_cleanup(
+    struct wsfs_message_queue * queue);
+
+extern bool wsfs_message_queue_empty(
+    struct wsfs_message_queue * queue);
+
+extern void wsfs_message_queue_push(
+    struct wsfs_message_queue * queue,
+    struct wsfs_message * message);
+
+extern struct wsfs_message * wsfs_message_queue_pop(
+    struct wsfs_message_queue * queue);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/wsfs/operation/close.c

@@ -4,27 +4,29 @@
 #include <errno.h>
 #include <jansson.h>
 
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
 #include "wsfs/util.h"
 
 
+static void wsfs_operation_close_finished(
+	void * user_data,
+	wsfs_status status,
+	json_t const * WSFS_UNUSED_PARAM(result))
+{
+	fuse_req_t request = (fuse_req_t) user_data;
+	fuse_reply_err(request, (WSFS_GOOD == status) ? 0 : ENOENT);
+}
+
 void wsfs_operation_close(
 	fuse_req_t request,
 	fuse_ino_t inode,
 	struct fuse_file_info * file_info)
 {
     struct wsfs_operations_context * user_data = fuse_req_userdata(request);
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
 
-	json_t * result = NULL;
 	int handle = (int) (file_info->fh & INT_MAX);
-	wsfs_status const status = wsfs_jsonrpc_invoke(rpc, &result, "close", "iii", inode, handle, file_info->flags);
+	wsfs_jsonrpc_server_invoke(
-	if (NULL != result)
+		rpc, &wsfs_operation_close_finished, request,
-	{
+		"close", "iii", inode, handle, file_info->flags);
-		// unused
-		json_decref(result);
-	}
-
-
-	fuse_reply_err(request, (WSFS_GOOD == status) ? 0 : ENOENT);
 }

src/wsfs/operation/getattr.c

@@ -7,21 +7,26 @@
 #include <sys/stat.h>
 #include <unistd.h> 
 
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
+#include "wsfs/jsonrpc/util.h"
 #include "wsfs/util.h"
 
-extern void wsfs_operation_getattr (
+struct wsfs_operation_getattr_context
-	fuse_req_t request,
-	fuse_ino_t inode,
-	struct fuse_file_info * WSFS_UNUSED_PARAM(file_info))
 {
-    struct fuse_ctx const * context = fuse_req_ctx(request);
+	fuse_req_t request;
-    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+	double timeout;
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
+	uid_t uid;
+	gid_t gid;
+};
+
+static void wsfs_operation_getattr_finished(
+	void * user_data,
+	wsfs_status status,
+	json_t const * data)
+{
+	struct wsfs_operation_getattr_context * context = user_data;
 
     struct stat buffer;
-	json_t * data = NULL;
-	wsfs_status status = wsfs_jsonrpc_invoke(rpc, &data, "getattr", "i", inode);
 	if (NULL != data)
 	{
 		json_t * mode_holder = json_object_get(data, "mode");
@@ -55,16 +60,34 @@ extern void wsfs_operation_getattr (
 		{
 	        status = WSFS_BAD_FORMAT;
 		}
-
-		json_decref(data);
 	}
 
     if (WSFS_GOOD == status)
     {
-        fuse_reply_attr(request, &buffer, user_data->timeout);
+        fuse_reply_attr(context->request, &buffer, context->timeout);
     }
     else
     {
-	    fuse_reply_err(request, ENOENT);
+	    fuse_reply_err(context->request, ENOENT);
     }
+
+	free(context);
+}
+
+void wsfs_operation_getattr (
+	fuse_req_t request,
+	fuse_ino_t inode,
+	struct fuse_file_info * WSFS_UNUSED_PARAM(file_info))
+{
+    struct fuse_ctx const * context = fuse_req_ctx(request);
+    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
+
+	struct wsfs_operation_getattr_context * getattr_context = malloc(sizeof(struct wsfs_operation_getattr_context));
+	getattr_context->request = request;
+	getattr_context->uid = context->uid;
+	getattr_context->gid = context->gid;
+	getattr_context->timeout = user_data->timeout;
+
+	wsfs_jsonrpc_server_invoke(rpc, &wsfs_operation_getattr_finished, getattr_context, "getattr", "i", inode);
 }

src/wsfs/operation/lookup.c

@@ -1,5 +1,6 @@
 #include "wsfs/operations.h"
 
+#include <limits.h>
 #include <errno.h>
 #include <string.h>
 
@@ -7,21 +8,29 @@
 #include <sys/stat.h>
 #include <unistd.h> 
 
-#include "wsfs/jsonrpc.h"
+#include <stdlib.h>
+
+#include "wsfs/jsonrpc/server.h"
+#include "wsfs/jsonrpc/util.h"
 #include "wsfs/util.h"
 
-void wsfs_operation_lookup (
+struct wsfs_operation_lookup_context
-	fuse_req_t request, 
-	fuse_ino_t parent, 
-	char const * name)
 {
-    struct fuse_ctx const * context = fuse_req_ctx(request);
+	fuse_req_t request;
-    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+	double timeout;
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
+	uid_t uid;
+	gid_t gid;
+};
 
+static void wsfs_operation_lookup_finished(
+	void * user_data,
+	wsfs_status status,
+	json_t const * data
+)
+{
+	struct wsfs_operation_lookup_context * context = user_data; 	
     struct fuse_entry_param buffer;
-	json_t * data = NULL;
+
-	wsfs_status status = wsfs_jsonrpc_invoke(rpc, &data, "lookup", "is", parent, name);
 	if (NULL != data)
 	{
 		json_t * inode_holder = json_object_get(data, "inode"); 
@@ -46,8 +55,8 @@ void wsfs_operation_lookup (
 			}
 
 
-			buffer.attr_timeout = user_data->timeout;
+			buffer.attr_timeout = context->timeout;
-			buffer.entry_timeout = user_data->timeout;
+			buffer.entry_timeout = context->timeout;
             buffer.attr.st_uid = context->uid;
             buffer.attr.st_gid = context->gid;
             buffer.attr.st_nlink = 1;
@@ -60,16 +69,35 @@ void wsfs_operation_lookup (
 		{
 	        status = WSFS_BAD_FORMAT;
 		}
-
-		json_decref(data);
 	}
 
     if (WSFS_GOOD == status)
     {
-        fuse_reply_entry(request, &buffer);
+        fuse_reply_entry(context->request, &buffer);
     }
     else
     {
-	    fuse_reply_err(request, ENOENT);
+	    fuse_reply_err(context->request, ENOENT);
     }
+
+	free(context);
+}
+
+void wsfs_operation_lookup (
+	fuse_req_t request, 
+	fuse_ino_t parent, 
+	char const * name)
+{
+    struct fuse_ctx const * context = fuse_req_ctx(request);
+    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
+
+	struct wsfs_operation_lookup_context * lookup_context = malloc(sizeof(struct wsfs_operation_lookup_context));
+	lookup_context->request = request;
+	lookup_context->uid = context->uid;
+	lookup_context->gid = context->gid;
+	lookup_context->timeout = user_data->timeout;
+
+	wsfs_jsonrpc_server_invoke(rpc, &wsfs_operation_lookup_finished, lookup_context, "lookup", "is", (int) (parent & INT_MAX), name);
+
 }

src/wsfs/operation/open.c

@@ -1,42 +1,52 @@
 #include "wsfs/operations.h"
 
+#include <string.h>
 #include <errno.h>
 #include <jansson.h>
 
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
 #include "wsfs/util.h"
 
+static void wsfs_operation_open_finished(
+	void * user_data,
+	wsfs_status status,
+	json_t const * result)
+{
+	fuse_req_t request = user_data;
+	struct fuse_file_info file_info;
+	memset(&file_info, 0, sizeof(struct fuse_file_info));
+
+	if (NULL != result)
+	{
+        json_t * handle_holder = json_object_get(result, "handle");
+        if ((NULL != handle_holder) && (json_is_integer(handle_holder))) 
+        {
+            file_info.fh = json_integer_value(handle_holder);
+        }
+        else
+        {
+            status = WSFS_BAD_FORMAT;
+        }        
+	}
+
+	if (WSFS_GOOD == status)
+	{
+		fuse_reply_open(request, &file_info);		
+	}
+	else
+	{
+		fuse_reply_err(request, ENOENT);
+	}
+
+}
+
 void wsfs_operation_open(
 	fuse_req_t request,
 	fuse_ino_t inode,
 	struct fuse_file_info * file_info)
 {
     struct wsfs_operations_context * user_data = fuse_req_userdata(request);
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
 
-	json_t * result = NULL;
+	wsfs_jsonrpc_server_invoke(rpc, &wsfs_operation_open_finished, request, "open", "ii", inode, file_info->flags);
-	wsfs_status status = wsfs_jsonrpc_invoke(rpc, &result, "open", "ii", inode, file_info->flags);
-	if (NULL != result)
-	{
-        json_t * handle_holder = json_object_get(result, "handle");
-        if ((NULL != handle_holder) && (json_is_integer(handle_holder))) 
-        {
-            file_info->fh = json_integer_value(handle_holder);
-        }
-        else
-        {
-            status = WSFS_BAD_FORMAT;
-        }        
-
-		json_decref(result);
-	}
-
-	if (WSFS_GOOD == status)
-	{
-		fuse_reply_open(request, file_info);		
-	}
-	else
-	{
-		fuse_reply_err(request, ENOENT);
-	}
 }

src/wsfs/operation/read.c

@@ -5,7 +5,7 @@
 #include <limits.h>
 #include <jansson.h>
 
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
 
 #define WSFS_MAX_READ_LENGTH 4096
 
@@ -34,22 +34,12 @@ static wsfs_status wsfs_fill_buffer(
 	return status;
 }
 
-void wsfs_operation_read(
+static void wsfs_operation_read_finished(void * user_data, wsfs_status status, json_t const * data)
-	fuse_req_t request,
-	fuse_ino_t inode,
-    size_t size,
-    off_t offset,
-	struct fuse_file_info * file_info)
 {
-    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+	fuse_req_t request = user_data;
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
 
-	int const length = (size <= WSFS_MAX_READ_LENGTH) ? (int) size : WSFS_MAX_READ_LENGTH;
+	char * buffer = NULL;	
-	char * buffer = malloc(length);
+	size_t length;
-	size_t count = 0;
-	json_t * data = NULL;
-    int handle = (file_info->fh & INT_MAX);
-	wsfs_status status = wsfs_jsonrpc_invoke(rpc, &data, "read", "iiii", inode, handle, (int) offset, length);
 	if (NULL != data)
 	{
 		json_t * data_holder = json_object_get(data, "data");
@@ -62,21 +52,20 @@ void wsfs_operation_read(
 		{
 			char const * const data = json_string_value(data_holder);
 			char const * const format = json_string_value(format_holder);
-			count = (size_t) json_integer_value(count_holder);
+			length = (size_t) json_integer_value(count_holder);
+			buffer = malloc(length);
 
-			status = wsfs_fill_buffer(buffer, length, format, data, count);
+			status = wsfs_fill_buffer(buffer, length, format, data, length);
 		}
 		else
 		{
 			status = WSFS_BAD_FORMAT;
 		}
-
-		json_decref(data);
 	}
 
 	if (WSFS_GOOD == status)
 	{
-		fuse_reply_buf(request, buffer, count);
+		fuse_reply_buf(request, buffer, length);
 	}
 	else
 	{
@@ -84,4 +73,20 @@ void wsfs_operation_read(
 	}
 	
     free(buffer);
+
+}
+
+void wsfs_operation_read(
+	fuse_req_t request,
+	fuse_ino_t inode,
+    size_t size,
+    off_t offset,
+	struct fuse_file_info * file_info)
+{
+    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
+
+	int const length = (size <= WSFS_MAX_READ_LENGTH) ? (int) size : WSFS_MAX_READ_LENGTH;
+    int handle = (file_info->fh & INT_MAX);
+	wsfs_jsonrpc_server_invoke(rpc, &wsfs_operation_read_finished, request, "read", "iiii", inode, handle, (int) offset, length);
 }

src/wsfs/operation/readdir.c

@@ -8,12 +8,19 @@
 #include <sys/stat.h>
 #include <unistd.h> 
 
-#include "wsfs/jsonrpc.h"
+#include "wsfs/jsonrpc/server.h"
 #include "wsfs/util.h"
 
 
 #define WSFS_DIRBUFFER_INITIAL_SIZE 1024
 
+struct wsfs_operation_readdir_context
+{
+	fuse_req_t request;
+	size_t size;
+	off_t offset;
+};
+
 struct wsfs_dirbuffer
 {
 	char * data;
@@ -64,21 +71,16 @@ static size_t min(size_t a, size_t b)
 	return (a < b) ? a : b;
 }
 
-void wsfs_operation_readdir (
+static void wsfs_operation_readdir_finished(
-	fuse_req_t request,
+	void * user_data,
-	fuse_ino_t inode,
+	wsfs_status status,
-	size_t size,
+	json_t const * result)
-	off_t offset,
-	struct fuse_file_info * WSFS_UNUSED_PARAM(file_info))
 {
-    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+	struct wsfs_operation_readdir_context * context = user_data;
-    struct wsfs_jsonrpc * rpc = user_data->rpc;
 
 	struct wsfs_dirbuffer buffer;
 	wsfs_dirbuffer_init(&buffer);
 
-	json_t * result = NULL;
-	wsfs_status const status = wsfs_jsonrpc_invoke(rpc, &result, "readdir", "i", inode);
 	if (NULL != result)
 	{
 		if (json_is_array(result)) 
@@ -98,32 +100,48 @@ void wsfs_operation_readdir (
 					{
 						char const * name = json_string_value(name_holder);
 						fuse_ino_t entry_inode = (fuse_ino_t) json_integer_value(inode_holder);
-						wsfs_dirbuffer_add(request, &buffer, name, entry_inode);	
+						wsfs_dirbuffer_add(context->request, &buffer, name, entry_inode);	
 					}
 				}
 			}
 		}
-
-		json_decref(result);
 	}
 
 	if (WSFS_GOOD == status)
 	{
-		if (((size_t) offset) < buffer.position)
+		if (((size_t) context->offset) < buffer.position)
 		{
-			fuse_reply_buf(request, &buffer.data[offset],
+			fuse_reply_buf(context->request, &buffer.data[context->offset],
-				min(buffer.position - offset, size));
+				min(buffer.position - context->offset, context->size));
 		}
 		else
 		{
-			fuse_reply_buf(request, NULL, 0);			
+			fuse_reply_buf(context->request, NULL, 0);			
 		}
 		
 	}
 	else
 	{
-		fuse_reply_err(request, ENOENT);
+		fuse_reply_err(context->request, ENOENT);
 	}
 
 	wsfs_dirbuffer_dispose(&buffer);
+	free(context);
+}
+
+void wsfs_operation_readdir (
+	fuse_req_t request,
+	fuse_ino_t inode,
+	size_t size,
+	off_t offset,
+	struct fuse_file_info * WSFS_UNUSED_PARAM(file_info))
+{
+    struct wsfs_operations_context * user_data = fuse_req_userdata(request);
+    struct wsfs_jsonrpc_server * rpc = user_data->rpc;
+	struct wsfs_operation_readdir_context * readdir_context = malloc(sizeof(struct wsfs_operation_readdir_context));
+	readdir_context->request = request;
+	readdir_context->size = size;
+	readdir_context->offset = offset;
+
+	 wsfs_jsonrpc_server_invoke(rpc, &wsfs_operation_readdir_finished, readdir_context, "readdir", "i", inode);
 }

src/wsfs/operations.h

@@ -3,11 +3,11 @@
 
 #include "wsfs/fuse_wrapper.h"
 
-struct wsfs_jsonrpc;
+struct wsfs_jsonrpc_server;
 
 struct wsfs_operations_context
 {
-	struct wsfs_jsonrpc * rpc;
+	struct wsfs_jsonrpc_server * rpc;
 	double timeout;
 };
 

src/wsfs/protocol.c

@@ -1,243 +0,0 @@
-#include "wsfs/protocol.h"
-
-#include <stdio.h>
-#include <stdlib.h>
-#include <stdbool.h>
-
-#include <pthread.h>
-
-#include "wsfs/util.h"
-#include "wsfs/server.h"
-
-struct wsfs_message
-{
-	char * content;
-	size_t length;
-};
-
-struct wsfs_protocol
-{
-	pthread_mutex_t lock;
-	struct lws * wsi;
-	struct wsfs_message pending_message;
-	struct wsfs_server * server;
-};
-
-static struct wsfs_protocol * wsfs_protocol_from_wsi(
-	struct lws * wsi)
-{
-	struct lws_protocols const * protocol = lws_get_protocol(wsi);
-	return protocol->user;
-}
-
-static bool wsfs_protocol_connect(
-	struct wsfs_protocol * protocol,
-	struct lws * wsi)
-{
-	pthread_mutex_lock(&protocol->lock);
-
-	bool const success = (NULL == protocol->wsi);
-	if (success)
-	{
-		protocol->wsi = wsi;
-	}
-
-	pthread_mutex_unlock(&protocol->lock);
-
-	return success;
-}
-
-static bool wsfs_protocol_is_wsi_connected(
-	struct wsfs_protocol * protocol,
-	struct lws * wsi)
-{
-	pthread_mutex_lock(&protocol->lock);
-	bool const result = (wsi == protocol->wsi);
-	pthread_mutex_unlock(&protocol->lock);
-
-	return result;
-}
-
-static void wsfs_protocol_disconnect(
-	struct wsfs_protocol * protocol,
-	struct lws * wsi)
-{
-	pthread_mutex_lock(&protocol->lock);
-	if (wsi == protocol->wsi)
-	{
-		protocol->wsi = NULL;
-	}
-	pthread_mutex_unlock(&protocol->lock);
-}
-
-static void wsfs_protocol_get_message(
-	struct wsfs_protocol * protocol, 
-	struct wsfs_message * message)
-{
-	pthread_mutex_lock(&protocol->lock);
-
-	message->content = protocol->pending_message.content;
-	message->length = protocol->pending_message.length;
-	
-	protocol->pending_message.content = NULL;
-	protocol->pending_message.length = 0;
-
-	pthread_mutex_unlock(&protocol->lock);
-}
-
-
-static int wsfs_protocol_callback(
-	struct lws *wsi,
-	enum lws_callback_reasons reason,
-	void * WSFS_UNUSED_PARAM(user),
-	void *in,
-	size_t len)
-{
-	int result = 0;
-	struct wsfs_protocol * const protocol = wsfs_protocol_from_wsi(wsi);
-	
-	switch (reason)
-	{
-		case LWS_CALLBACK_ESTABLISHED:
-		{
-			if (!wsfs_protocol_connect(protocol, wsi))
-			{
-				puts("connect failed");
-				lws_callback_on_writable(wsi);
-				result = -1;
-			}
-		}
-		break;
-		case LWS_CALLBACK_CLOSED:
-		{
-			wsfs_protocol_disconnect(protocol, wsi);
-		}
-		break;
-		case LWS_CALLBACK_SERVER_WRITEABLE:
-		{
-			if (wsfs_protocol_is_wsi_connected(protocol, wsi))
-			{
-				struct wsfs_message message;
-				wsfs_protocol_get_message(protocol, &message);
-				if (NULL != message.content)
-				{
-					lws_write(wsi, (unsigned char*) message.content, message.length, LWS_WRITE_TEXT);
-					wsfs_protocol_message_dispose(message.content);
-				}
-			}
-			else
-			{
-				result = -1;
-			}
-		}
-		break;
-		case LWS_CALLBACK_RECEIVE:
-		{
-			wsfs_server_handle_message(protocol->server, in, len);
-		}
-		break;
-		default:
-		break;
-	}
-
-	return result;
-}
-
-
-struct wsfs_protocol * wsfs_protocol_create(
-	struct wsfs_server * server)
-{
-	struct wsfs_protocol * protocol = malloc(sizeof(struct wsfs_protocol));
-	if (NULL != protocol)
-	{	
-		pthread_mutex_init(&protocol->lock, NULL);
-		protocol->wsi = NULL;
-		protocol->pending_message.content = NULL;
-		protocol->pending_message.length = 0;
-		protocol->server = server;
-	}
-
-	return protocol;
-}
-
-void wsfs_protocol_dispose(
-	struct wsfs_protocol * protocol)
-{
-	pthread_mutex_destroy(&protocol->lock);
-
-	if (NULL != protocol->pending_message.content)
-	{
-		wsfs_protocol_message_dispose(protocol->pending_message.content);
-		protocol->pending_message.content = NULL;
-	}
-
-	free(protocol);
-}
-
-void wsfs_protocol_check(
-	struct wsfs_protocol * protocol)
-{
-	pthread_mutex_lock(&protocol->lock);
-
-	if ((NULL != protocol->wsi) && (NULL != protocol->pending_message.content))
-	{
-		lws_callback_on_writable(protocol->wsi);
-	}
-
-	pthread_mutex_unlock(&protocol->lock);
-}
-
-void wsfs_protocol_init_lws(
-	struct wsfs_protocol * protocol,
-	struct lws_protocols * lws_protocol)
-{
-	lws_protocol->callback = &wsfs_protocol_callback;
-	lws_protocol->per_session_data_size = 1;
-	lws_protocol->user = protocol;
-}
-
-char * wsfs_protocol_message_create(
-	size_t size)
-{
-	char * buffer = malloc(LWS_PRE + size);
-	return &buffer[LWS_PRE];
-}
-
-void wsfs_protocol_message_dispose(
-	char * message)
-{
-	char * buffer = message - LWS_PRE;
-	free(buffer);
-}
-
-bool wsfs_protocol_send(
-	char * message,
-	size_t length,
-	void * user_data)
-{
-	struct wsfs_protocol * protocol = user_data;
-
-	pthread_mutex_lock(&protocol->lock);
-
-	struct wsfs_server * server = protocol->server;
-	bool result = (NULL != protocol->wsi) && (NULL == protocol->pending_message.content);
-	if (result)
-	{
-		protocol->pending_message.content = message;
-		protocol->pending_message.length = length;
-	}
-	else
-	{
-		wsfs_protocol_message_dispose(message);
-	}
-
-	pthread_mutex_unlock(&protocol->lock);	
-
-	if (result)
-	{
-		wsfs_server_wakeup(server);
-	}
-
-	return result;
-}
-

src/wsfs/protocol.h

@@ -1,48 +0,0 @@
-#ifndef _WSFS_PROTOCOL_H
-#define _WSFS_PROTOCOL_H
-
-#ifndef __cplusplus
-#include <stdbool.h>
-#endif
-
-#include <libwebsockets.h>
-#include <pthread.h>
-
-struct wsfs_protocol;
-struct wsfs_server;
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-extern struct wsfs_protocol * wsfs_protocol_create(
-	struct wsfs_server * server);
-
-extern void wsfs_protocol_dispose(
-	struct wsfs_protocol * protocol);
-
-extern void wsfs_protocol_check(
-	struct wsfs_protocol * protocol);
-
-extern void wsfs_protocol_init_lws(
-	struct wsfs_protocol * protocl,
-	struct lws_protocols * lws_protocol);
-
-extern char * wsfs_protocol_message_create(
-	size_t size);
-
-extern void wsfs_protocol_message_dispose(
-	char * message);
-
-extern bool wsfs_protocol_send(
-	char * message,
-	size_t length,
-	void * user_data);
-
-#ifdef __cplusplus
-}
-#endif
-
-
-#endif
-

src/wsfs/response_parser.h

@@ -1,35 +0,0 @@
-#ifndef _WSFS_RESPONSE_PARSER_H
-#define _WFSF_RESPONSE_PARSER_H
-
-#ifndef __cplusplus
-#include <stddef.h>
-#else
-#include <cstddef>
-using std::size_t;
-#endif
-
-#include <jansson.h>
-#include "wsfs/status.h"
-
-struct wsfs_response
-{
-	wsfs_status status;
-	int id;
-	json_t * result;	
-};
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-extern void wsfs_response_parse(
-	char const * buffer,
-	size_t buffer_length,
-	struct wsfs_response * response);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif
-

src/wsfs/server.c

@@ -17,6 +17,8 @@ struct wsfs_server
     struct lws_protocols ws_protocols[WSFS_SERVER_PROTOCOL_COUNT];
     struct lws_context * context;
     volatile bool shutdown_requested;
+	struct lws_http_mount mount;
+	struct lws_context_creation_info info;
 };
 
 static bool wsfs_server_tls_enabled(
@@ -34,50 +36,37 @@ static struct lws_context * wsfs_server_context_create(
     server->ws_protocols[1].name = "fs";
     wsfs_server_protocol_init_lws(&server->protocol, &server->ws_protocols[1]);
 
-	struct lws_http_mount mount = 
+	memset(&server->mount, 0, sizeof(struct lws_http_mount));
-	{
+	server->mount.mount_next = NULL,
-		.mount_next = NULL,
+	server->mount.mountpoint = "/",
-		.mountpoint = "/",
+	server->mount.origin = server->config.document_root,
-		.origin = server->config.document_root,
+	server->mount.def = "index.html",
-		.def = "index.html",
+	server->mount.origin_protocol = LWSMPRO_FILE,
-		.protocol = NULL,
+	server->mount.mountpoint_len = 1,
-		.cgienv = NULL,
-		.extra_mimetypes = NULL,
-		.interpret = NULL,
-		.cgi_timeout = 0,
-		.cache_max_age = 0,
-		.auth_mask = 0,
-		.cache_reusable = 0,
-		.cache_intermediaries = 0,
-		.origin_protocol = LWSMPRO_FILE,
-		.mountpoint_len = 1,
-		.basic_auth_login_file = NULL
-	};
 
-	struct lws_context_creation_info info;
+	memset(&server->info, 0, sizeof(struct lws_context_creation_info));
-	memset(&info, 0, sizeof(info));
+	server->info.port = server->config.port;
-	info.port = server->config.port;
+	server->info.mounts = &server->mount;
-	info.mounts = &mount;
+	server->info.protocols = server->ws_protocols;
-	info.protocols = server->ws_protocols;
+	server->info.vhost_name = server->config.vhost_name;
-	info.vhost_name = server->config.vhost_name;
+	server->info.ws_ping_pong_interval = 10;
-	info.ws_ping_pong_interval = 10;
+	server->info.options = LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;
-	info.options = LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;
 
 	if (NULL == server->config.document_root)
 	{
 		// disable http
-		info.protocols = &server->ws_protocols[1];
+		server->info.protocols = &server->ws_protocols[1];
-		info.mounts = NULL;
+		server->info.mounts = NULL;
 	}
 
 	if (wsfs_server_tls_enabled(server))
 	{
-		info.options |= LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
+		server->info.options |= LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
-		info.ssl_cert_filepath = server->config.cert_path;
+		server->info.ssl_cert_filepath = server->config.cert_path;
-		info.ssl_private_key_filepath = server->config.key_path;
+		server->info.ssl_private_key_filepath = server->config.key_path;
 	}
 
-	struct lws_context * const context = lws_create_context(&info);
+	struct lws_context * const context = lws_create_context(&server->info);
     return context;
 
 }
@@ -87,12 +76,19 @@ struct wsfs_server * wsfs_server_create(
 {
     struct wsfs_server * server = malloc(sizeof(struct wsfs_server));
     if (NULL != server)
+    {
+        if (wsfs_server_protocol_init(&server->protocol, config->mount_point))
 		{
 			server->shutdown_requested = false;
 			wsfs_server_config_clone(config, &server->config);
-        wsfs_server_protocol_init(&server->protocol, config->mount_point);
 			server->context = wsfs_server_context_create(server);
 		}
+		else
+		{
+			free(server);
+			server = NULL;
+		}
+    }   
 
     return server; 
 }

src/wsfs/server_protocol.c

@@ -3,6 +3,7 @@
 #include <stdlib.h>
 #include <libwebsockets.h>
 
+#include "wsfs/message.h"
 #include "wsfs/filesystem.h"
 #include "wsfs/util.h"
 
@@ -27,26 +28,91 @@ static int wsfs_server_protocol_callback(
                 puts("error: unable to adopt fd");
             }
         }
+        break;
+		case LWS_CALLBACK_ESTABLISHED:
+		{
+			if (NULL == protocol->wsi)
+			{
+                protocol->wsi = wsi;
+			}
+		}
+		break;
+		case LWS_CALLBACK_CLOSED:
+		{
+			if (wsi == protocol->wsi)
+            {
+                protocol->wsi = NULL;
+                wsfs_message_queue_cleanup(&protocol->queue);
+            }
+		}
+        break;
+		case LWS_CALLBACK_SERVER_WRITEABLE:
+		{
+			if ((wsi == protocol->wsi) && (!wsfs_message_queue_empty(&protocol->queue)))
+			{                
+				struct wsfs_message * message = wsfs_message_queue_pop(&protocol->queue);
+				lws_write(wsi, (unsigned char*) message->data, message->length, LWS_WRITE_TEXT);
+				wsfs_message_dispose(message);
+			}
+		}
 		break;
         case LWS_CALLBACK_RAW_RX_FILE:
         {
-            puts("fd readable");
             wsfs_filesystem_process_request(&protocol->filesystem);
         }
         break;
         default:
         break;
     }
+
+    if ((wsi == protocol->wsi) && (!wsfs_message_queue_empty(&protocol->queue)))
+    {
+        lws_callback_on_writable(wsi);
+    }
+
     return 0;
 }
 
+static bool wsfs_server_protocol_invoke(
+    void * user_data,
+    json_t const * request)
+{
+    bool result = false;
+    struct wsfs_server_protocol * protocol = user_data;
+
+    if (NULL != protocol->wsi)
+    {
+        size_t length = json_dumpb(request, NULL, 0, JSON_COMPACT);
+        if (0 < length)
+        {	
+            struct wsfs_message * message = wsfs_message_create(length);
+            json_dumpb(request, message->data, length, JSON_COMPACT);
+
+            wsfs_message_queue_push(&protocol->queue, message);
+            lws_callback_on_writable(protocol->wsi);
+
+            // ToDo: add timeout
+
+            result = true;
+        }
+    }
+    
+
+    return result;
+}
+
+
 struct wsfs_server_protocol * wsfs_server_protocol_create(
     char * mount_point)
 {
     struct wsfs_server_protocol * protocol = malloc(sizeof(struct wsfs_server_protocol));
     if (NULL != protocol)
     {
-        wsfs_server_protocol_init(protocol, mount_point);
+        if (!wsfs_server_protocol_init(protocol, mount_point))
+        {
+            free(protocol);
+            protocol = NULL;
+        }
     }
 
     return protocol;
@@ -68,15 +134,29 @@ void wsfs_server_protocol_init_lws(
 	lws_protocol->user = protocol;
 }
 
-void wsfs_server_protocol_init(
+bool wsfs_server_protocol_init(
     struct wsfs_server_protocol * protocol,
     char * mount_point)
 {
-    wsfs_filesystem_init(&protocol->filesystem, mount_point);
+    protocol->wsi = NULL;
+    wsfs_message_queue_init(&protocol->queue);
+
+    wsfs_jsonrpc_server_init(&protocol->rpc);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "lookup", &wsfs_server_protocol_invoke, protocol);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "getattr", &wsfs_server_protocol_invoke, protocol);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "readdir", &wsfs_server_protocol_invoke, protocol);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "open", &wsfs_server_protocol_invoke, protocol);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "close", &wsfs_server_protocol_invoke, protocol);
+    wsfs_jsonrpc_server_add(&protocol->rpc, "read", &wsfs_server_protocol_invoke, protocol);
+
+    return wsfs_filesystem_init(&protocol->filesystem, &protocol->rpc, mount_point);
 }
 
 void wsfs_server_protocol_cleanup(
     struct wsfs_server_protocol * protocol)
 {
     wsfs_filesystem_cleanup(&protocol->filesystem);
+    wsfs_jsonrpc_server_cleanup(&protocol->rpc);
+    wsfs_message_queue_cleanup(&protocol->queue);
+    protocol->wsi = NULL;
 }

src/wsfs/server_protocol_intern.h

@@ -3,13 +3,18 @@
 
 #include "wsfs/server_protocol.h"
 #include "wsfs/filesystem.h"
+#include "wsfs/message_queue.h"
+#include "wsfs/jsonrpc/server.h"
 
 struct wsfs_server_protocol
 {
     struct wsfs_filesystem filesystem;
+    struct wsfs_jsonrpc_server rpc;
+    struct wsfs_message_queue queue;
+    struct lws * wsi;    
 };
 
-extern void wsfs_server_protocol_init(
+extern bool wsfs_server_protocol_init(
     struct wsfs_server_protocol * protocol,
     char * mount_point);
 

src/wsfs/status.c

@@ -7,6 +7,9 @@
 	switch(status)
 	{
 		case WSFS_GOOD: return 0;
+		case WSFS_BAD_NOTIMPLEMENTED: return -ENOSYS;
+		case WSFS_BAD_TIMEOUT: return -ETIMEDOUT;
+		case WSFS_BAD_BUSY: return -ENOENT;
 		case WSFS_BAD_NOENTRY: return -ENOENT;
 		case WSFS_BAD_NOACCESS: return -EACCES;
 		default: return -ENOENT;

src/wsfs/status.h

@@ -6,7 +6,8 @@
 
 #define WSFS_BAD_NOTIMPLEMENTED 2
 #define WSFS_BAD_TIMEOUT        3
-#define WSFS_BAD_FORMAT         4
+#define WSFS_BAD_BUSY           4
+#define WSFS_BAD_FORMAT         5
 
 #define WSFS_BAD_NOENTRY 101
 #define WSFS_BAD_NOACCESS 102

test-src/test_fuse_req.cc

@@ -0,0 +1,7 @@
+#include <gtest/gtest.h>
+#include "wsfs/fuse_wrapper.h"
+
+TEST(libfuse, fuse_req_t_size)
+{
+    ASSERT_EQ(sizeof(void*), sizeof(fuse_req_t));
+}

test-src/test_response_parser.cc

@@ -2,20 +2,20 @@
 #include <gtest/gtest.h>
 
 extern "C" {
-#include "wsfs/response_parser.h"
+#include "wsfs/jsonrpc/response.h"
 }
 
 static void wsfs_response_parse_str(
 	char const * buffer,
-	struct wsfs_response * response)
+	struct wsfs_jsonrpc_response * response)
 {
 	size_t length = strlen(buffer);
-	wsfs_response_parse(buffer, length, response);
+	wsfs_jsonrpc_response_init(response, buffer, length);
 }
 
 TEST(response_parser, test)
 {
-	struct wsfs_response response;
+	struct wsfs_jsonrpc_response response;
 
 	// invalid json
 	wsfs_response_parse_str("", &response);

test-src/test_server.cc

@@ -1,11 +1,24 @@
 #include <gtest/gtest.h>
+
+#include <cstdlib>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
 #include "wsfs/server.h"
+#include "wsfs/server_config.h"
+
 
 TEST(server, create_dispose)
 {
-    wsfs_server_config config = {nullptr, nullptr, nullptr, nullptr, 0};
+    mkdir("test", 0700);
-    wsfs_server * server = wsfs_server_create(&config);
+
+    struct wsfs_server_config config = {strdup("test"), nullptr, nullptr, nullptr, nullptr, 0};
+    struct wsfs_server * server = wsfs_server_create(&config);
     ASSERT_NE(nullptr, server);
 
     wsfs_server_dispose(server);
+    wsfs_server_config_cleanup(&config);
+
+    rmdir("test");
 }