diff options
| -rw-r--r-- | src/connectionhandler.cc | 2 | ||||
| -rw-r--r-- | src/munia_proto.cc | 116 |
2 files changed, 92 insertions, 26 deletions
diff --git a/src/connectionhandler.cc b/src/connectionhandler.cc index c8881a3..da94cb8 100644 --- a/src/connectionhandler.cc +++ b/src/connectionhandler.cc @@ -67,7 +67,7 @@ void ConnectionHandler::unobserve(clientid_t clientid, taskid_t taskid) ObserverList ConnectionHandler::observerlist(TaskIdList tasks) { - printf("Observerlist request\n"); + printf("Observerlist request (#tasks: %d)\n", tasks.size()); ObserverList clients; for(TaskIdList::iterator i = tasks.begin(); i != tasks.end(); i++) { diff --git a/src/munia_proto.cc b/src/munia_proto.cc index a91c204..b7123aa 100644 --- a/src/munia_proto.cc +++ b/src/munia_proto.cc @@ -81,7 +81,7 @@ static void dump_handshake_info(struct lws_tokens *lwst) } #endif -static std::map<struct libwebsocket *, std::queue<std::string> > msgqueue; +static std::map<struct libwebsocket *, std::queue<message_t> > msgqueue; int callback_lws_task(struct libwebsocket_context * context, struct libwebsocket *wsi, @@ -102,11 +102,26 @@ 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) { - std::string msg = msgqueue[wsi].front(); + 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, @@ -127,6 +142,34 @@ int callback_lws_task(struct libwebsocket_context * context, libwebsocket_callback_on_writable(context, wsi); } } + */ + { + std::string msgstr; + msgstr.append((size_t)LWS_SEND_BUFFER_PRE_PADDING, ' '); + + while(msgqueue[wsi].size() > 0) { + message_t msg = msgqueue[wsi].front(); + msgqueue[wsi].pop(); + char buf[32]; + sprintf(buf, "%d", msg.tid); + 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); + if(n < 0) { + fprintf(stderr, "ERROR writing to socket"); + exit(1); + } + } + break; /* @@ -152,30 +195,53 @@ int callback_lws_task(struct libwebsocket_context * context, MessageList::iterator omi = omsgs.begin(); while(omi != omsgs.end()) { - std::string msgcmd = msg_tostring(*omi); - - printf("%d nodes affected by command\n", omi->nodes.size()); - - std::list<std::pair<clientid_t, taskid_t> > clients = - connection_handler.observerlist(omi->nodes); - printf("Writing message to %d clients\n", clients.size()); - - std::list<std::pair<clientid_t, taskid_t> >::iterator ci = clients.begin(); - while(ci != clients.end()) { - 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, ' '); + + if(omi->cmd == cmd::observe) { + + TaskIdList ids = task_manager.subTasks(omi->observe.id); + TaskIdList::iterator id = ids.begin(); + while(id != ids.end()) { + task_t task = task_manager.task(*id); + + message_t createmsg = create_msg_create(task); + message_t updatemsg = create_msg_update(task); + msgqueue[wsi].push(createmsg); + msgqueue[wsi].push(updatemsg); + + id++; + } + + } else if(omi->cmd == cmd::unobserve) { + + TaskIdList ids = task_manager.subTasks(omi->observe.id); + TaskIdList::iterator id = ids.begin(); + while(id != ids.end()) { + task_t task = task_manager.task(*id); - msgqueue[clientid].push(msg); - ci++; + message_t removemsg = create_msg_remove(task); + msgqueue[wsi].push(removemsg); + + id++; + } + + } else { + printf("%d nodes affected by command\n", omi->nodes.size()); + + ObserverList clients = connection_handler.observerlist(omi->nodes); + printf("Writing message to %d clients\n", clients.size()); + + ObserverList::iterator ci = clients.begin(); + while(ci != clients.end()) { + clientid_t clientid = (*ci).first; + taskid_t tid = (*ci).second; + + message_t msg = *omi; + msg.tid = tid; + + msgqueue[clientid].push(msg); + + ci++; + } } omi++; |