diff options
| -rw-r--r-- | src/munia_proto.cc | 136 |
1 files changed, 49 insertions, 87 deletions
diff --git a/src/munia_proto.cc b/src/munia_proto.cc index 3621bbf..7e63d73 100644 --- a/src/munia_proto.cc +++ b/src/munia_proto.cc @@ -81,7 +81,8 @@ static void dump_handshake_info(struct lws_tokens *lwst) } #endif -static std::map<struct libwebsocket *, std::queue<message_t> > msgqueue; +typedef std::list<message_t> msgqueue_t; +static std::map<struct libwebsocket *, msgqueue_t > msgqueue; int callback_lws_task(struct libwebsocket_context * context, struct libwebsocket *wsi, @@ -102,108 +103,68 @@ int callback_lws_task(struct libwebsocket_context * context, break; case LWS_CALLBACK_SERVER_WRITEABLE: - /* { - printf("Socket for client %p writable\n", wsi); - if(msgqueue[wsi].size() > 0) { - message_t msg = msgqueue[wsi].front(); - msgqueue[wsi].pop(); - - std::string msgcmd = msg_tostring(msg); - std::string msg; - clientid_t clientid = (*ci).first; - char tidstr[32]; - sprintf(tidstr, "%u", (*ci).second); - printf("\tAdding data to %p's queue\n", clientid); - - msg.append((size_t)LWS_SEND_BUFFER_PRE_PADDING, ' '); - msg.append(tidstr); - msg.append(" "); - msg.append(msgcmd); - msg.append((size_t)LWS_SEND_BUFFER_POST_PADDING, ' '); - - int n = libwebsocket_write(wsi, (unsigned char *) - msg.c_str() + - LWS_SEND_BUFFER_PRE_PADDING, - msg.length() - LWS_SEND_BUFFER_POST_PADDING - - LWS_SEND_BUFFER_PRE_PADDING, - LWS_WRITE_TEXT); - if(n < 0) { - fprintf(stderr, "ERROR writing to socket"); - exit(1); + // + // Find and collapse remove/create msg's into single move msg at + // the position of the remove msg. + // + msgqueue_t::iterator i = msgqueue[wsi].begin(); + while(i != msgqueue[wsi].end()) { + message_t &msg = *i; + if(msg.cmd == cmd::remove) { + + msgqueue_t::iterator j = msgqueue[wsi].begin(); + while(j != msgqueue[wsi].end()) { + message_t m = *j; + if(m.cmd == cmd::create && // msg was a create msg? + // create with same ids as we just removed? + msg.remove.id == m.create.id && msg.tid == m.tid) { + break; + } + j++; + } + + // collapse into move msg + if(j != msgqueue[wsi].end()) { + taskid_t id = j->create.id; + taskid_t parentid = j->create.id; + msgqueue[wsi].erase(j); // erase create msg + msg.cmd = cmd::move; + msg.move.id = id; + msg.move.parentid = parentid; + } } - } - else { - printf("\tNo queued data awaiting transmission\n"); + i++; } - if(msgqueue[wsi].size()) { - libwebsocket_rx_flow_control(wsi, 1); - libwebsocket_callback_on_writable(context, wsi); - } - } - */ - { + std::string msgstr; msgstr.append((size_t)LWS_SEND_BUFFER_PRE_PADDING, ' '); - int c = 0; - message_t prevmsg; - std::queue<message_t> messages; while(msgqueue[wsi].size() > 0) { message_t msg = msgqueue[wsi].front(); - msgqueue[wsi].pop(); - - //Rewriting delete id followed by create id to move to_id from_id - // look at previous message from create commands - if(c > 0 && // look backward (skip first entry) - msg.cmd == cmd::create && // look from create commands - prevmsg.cmd == cmd::remove && // if previous msg is a remove command - msg.create.id == prevmsg.remove.id) { // and commands have same id - - printf("\tprevmsg: %s, msg: %s\n", msg_tostring(prevmsg).c_str(), - msg_tostring(msg).c_str()); - message_t movemsg = create_msg_move(msg.create.id, msg.create.parentid); - movemsg.tid = msg.tid; - msg = movemsg; - } - else if(c > 0) { - messages.push(prevmsg); - } - - prevmsg = msg; - c++; - } - if(c > 0) { // Push last msg - messages.push(prevmsg); - } - - // create msg string - // while(msgqueue[wsi].size() > 0) { - while(messages.size() > 0) { - message_t msg = messages.front(); - messages.pop(); + msgqueue[wsi].pop_front(); char buf[32]; sprintf(buf, "%d", msg.tid); - printf("msg: %s\n", msg_tostring(msg).c_str()); if(msgstr.size() > LWS_SEND_BUFFER_PRE_PADDING) msgstr += " "; msgstr += std::string(buf) + " " + msg_tostring(msg); } msgstr.append((size_t)LWS_SEND_BUFFER_POST_PADDING, ' '); - + int n = libwebsocket_write(wsi, (unsigned char *) - msgstr.c_str() + - LWS_SEND_BUFFER_PRE_PADDING, - msgstr.length() - - LWS_SEND_BUFFER_POST_PADDING - - LWS_SEND_BUFFER_PRE_PADDING, - LWS_WRITE_TEXT); + msgstr.c_str() + + LWS_SEND_BUFFER_PRE_PADDING, + msgstr.length() - + LWS_SEND_BUFFER_POST_PADDING - + LWS_SEND_BUFFER_PRE_PADDING, + LWS_WRITE_TEXT); if(n < 0) { fprintf(stderr, "ERROR writing to socket"); exit(1); - } } - + } + } + break; /* @@ -242,8 +203,8 @@ int callback_lws_task(struct libwebsocket_context * context, message_t updatemsg = create_msg_update(task); updatemsg.tid = omi->observe.id; - msgqueue[wsi].push(createmsg); - msgqueue[wsi].push(updatemsg); + msgqueue[wsi].push_back(createmsg); + msgqueue[wsi].push_back(updatemsg); id++; } @@ -257,10 +218,11 @@ int callback_lws_task(struct libwebsocket_context * context, message_t removemsg = create_msg_remove(task); removemsg.tid = omi->observe.id; - msgqueue[wsi].push(removemsg); + msgqueue[wsi].push_back(removemsg); id++; } + } else { printf("%d nodes affected by command\n", omi->nodes.size()); @@ -277,7 +239,7 @@ int callback_lws_task(struct libwebsocket_context * context, message_t msg = *omi; msg.tid = tid; - msgqueue[clientid].push(msg); + msgqueue[clientid].push_back(msg); ci++; } |