Send a single message when WRITEABLE state (#237)

* Send a single message when WRITEABLE state

* Update libwebsockets-plugin.cpp

* Remove write_request

* Check write queue size before requesting writable callback

Change-Id: I89317d7fc8dc39ccc2eeb3e9a864940b7184228f

diff --git a/plugins/libwebsockets/libwebsockets-plugin.cpp b/plugins/libwebsockets/libwebsockets-plugin.cpp
index 1631b16..942d992 100755 (executable)
--- a/plugins/libwebsockets/libwebsockets-plugin.cpp
+++ b/plugins/libwebsockets/libwebsockets-plugin.cpp
@@ -308,10 +308,7 @@ static void _change_websocket_poll_fd(struct lws_pollargs *args)
 // For debug
 static void __print_received_data(void *in, size_t len)
 {
-       char *data = (char *)calloc(len, sizeof(char));
-       memcpy(data, in, len);
-       VINE_LOGD(">> len[%zd] received[%s]", len, data);
-       free(data);
+       VINE_LOGD("len[%zd] received [%s]", len, (char *)in);
 }
 
 static int _websocket_protocol_cb(struct lws *wsi,
@@ -443,7 +440,7 @@ static int _websocket_protocol_cb(struct lws *wsi,
                        return -1;
                }
 
-               if (g_callbacks.written_cb)
+               if (n && g_callbacks.written_cb)
                        g_callbacks.written_cb(n, ws->user);
                break;
 
@@ -731,12 +728,13 @@ static int _write_data(websocket_s *ws)
        if (!ws->write_buffer)
                return bytes;
 
-       while ((wd = ws->write_buffer->pop())) {
+       if ((wd = ws->write_buffer->pop())) {
                unsigned char *out = (unsigned char *)calloc(LWS_PRE + wd->len,
                                sizeof(unsigned char));
                memcpy(out + LWS_PRE, wd->buf, wd->len);
                int n = lws_write(ws->wsi, out + LWS_PRE, wd->len, LWS_WRITE_BINARY);
                if (n < 0) {
+                       VINE_LOGE("lws_write() is failed.");
                        free(out);
                        __destroy_websocket_data(wd);
                        return n;
@@ -749,6 +747,9 @@ static int _write_data(websocket_s *ws)
                wd = NULL;
        }
 
+       if (ws->write_buffer->size() > 0)
+               _request_write(ws);
+
        return bytes;
 }
 
@@ -815,7 +816,7 @@ static int websocket_write(vine_dp_plugin_h handle, unsigned char *buf, size_t l
        wd->len = len;
 
        ws->write_buffer->push(wd);
-       VINE_LOGD("websocket_data[%p] is pushed to write_buffer.", wd);
+       VINE_LOGD("websocket_data[%p] is pushed to write_buffer. len[%d]", wd, len);
 
        if (_add_websocket_op_request(WEBSOCKET_OP_WRITE, ws, 0, NULL, -1, NULL, 0, NULL) < 0)
                return VINE_DATA_PATH_ERROR_OPERATION_FAILED;

diff --git a/src/include/vine-queue.h b/src/include/vine-queue.h
index 681d357..e163127 100755 (executable)
--- a/src/include/vine-queue.h
+++ b/src/include/vine-queue.h
@@ -63,6 +63,12 @@ public:
                return _queue.front();
        }
 
+       size_t size()
+       {
+               std::lock_guard<std::mutex> lock_guard(_q_mutex);
+               return _queue.size();
+       }
+
 private:
        std::queue<T> _queue;
        std::mutex _q_mutex;

diff --git a/src/vine-event-loop.cpp b/src/vine-event-loop.cpp
index bff62cc..d4af0c7 100644 (file)
--- a/src/vine-event-loop.cpp
+++ b/src/vine-event-loop.cpp
@@ -284,7 +284,7 @@ int vine_event_loop_process(vine_event_queue_h event_fd)
                return VINE_ERROR_OPERATION_FAILED;
        }
 
-       VINE_LOGD("eventfd counter: %ld", u);
+       VINE_LOGD("eventfd counter: %lld", u);
 
        while (u--) {
                vine_event *event = event_fd_handle->event_queue.pop();