From fd373d0b0f580b56f3350e92207cfa499806ca4d Mon Sep 17 00:00:00 2001 From: Jonas Suhr Christensen Date: Thu, 12 Apr 2012 11:30:40 +0200 Subject: Added a bit of debugging messages. Merged task_proto and observe_proto into one protocol in munia_proto. --- src/munia_proto.cc | 188 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 src/munia_proto.cc (limited to 'src/munia_proto.cc') diff --git a/src/munia_proto.cc b/src/munia_proto.cc new file mode 100644 index 0000000..3679a81 --- /dev/null +++ b/src/munia_proto.cc @@ -0,0 +1,188 @@ +/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */ +/* vim: set et sw=2 ts=2: */ +/*************************************************************************** + * munia_proto.cc + * + * Fri Feb 24 08:23:16 CET 2012 + * Copyright 2012 Bent Bisballe Nyeng + * deva@aasimon.org + ****************************************************************************/ + +/* + * This file is part of Munia. + * + * Munia is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version. + * + * Munia is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with Munia; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. + */ +#include "munia_proto.h" + +#include +#include + +#include +#include + +#include "task.h" +#include "messageparser.h" +#include "messagehandler.h" +#include "connectionhandler.h" + +#include "taskmanager.h" + +extern TaskManager taskman; + +#if 0 +static void dump_handshake_info(struct lws_tokens *lwst) +{ + int n; + static const char *token_names[WSI_TOKEN_COUNT] = { + /*[WSI_TOKEN_GET_URI] =*/ "GET URI", + /*[WSI_TOKEN_HOST] =*/ "Host", + /*[WSI_TOKEN_CONNECTION] =*/ "Connection", + /*[WSI_TOKEN_KEY1] =*/ "key 1", + /*[WSI_TOKEN_KEY2] =*/ "key 2", + /*[WSI_TOKEN_PROTOCOL] =*/ "Protocol", + /*[WSI_TOKEN_UPGRADE] =*/ "Upgrade", + /*[WSI_TOKEN_ORIGIN] =*/ "Origin", + /*[WSI_TOKEN_DRAFT] =*/ "Draft", + /*[WSI_TOKEN_CHALLENGE] =*/ "Challenge", + + /* new for 04 */ + /*[WSI_TOKEN_KEY] =*/ "Key", + /*[WSI_TOKEN_VERSION] =*/ "Version", + /*[WSI_TOKEN_SWORIGIN] =*/ "Sworigin", + + /* new for 05 */ + /*[WSI_TOKEN_EXTENSIONS] =*/ "Extensions", + + /* client receives these */ + /*[WSI_TOKEN_ACCEPT] =*/ "Accept", + /*[WSI_TOKEN_NONCE] =*/ "Nonce", + /*[WSI_TOKEN_HTTP] =*/ "Http", + /*[WSI_TOKEN_MUXURL] =*/ "MuxURL", + }; + + for (n = 0; n < WSI_TOKEN_COUNT; n++) { + if (lwst[n].token == NULL || lwst[n].token_len == 0) continue; + fprintf(stderr, " %s = ", token_names[n]); + if(fwrite(lwst[n].token, 1, lwst[n].token_len, stderr)) {} + fprintf(stderr, "\n"); + } +} +#endif + +static std::map > msgqueue; + +int callback_lws_task(struct libwebsocket_context * context, + struct libwebsocket *wsi, + enum libwebsocket_callback_reasons reason, + void *user, void *in, size_t len) +{ + taskman.tree.toStdOut(); + + printf("Callback on %p\n", wsi); + + switch (reason) { + case LWS_CALLBACK_ESTABLISHED: + connection_handler.init(wsi); + break; + + case LWS_CALLBACK_CLOSED: + connection_handler.close(wsi); + 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(); + msgqueue[wsi].pop(); + 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); + } + } + else { + printf("\tNo queued data awaiting transmission\n"); + } + + if(msgqueue[wsi].size()) { + libwebsocket_rx_flow_control(wsi, 1); + libwebsocket_callback_on_writable(context, wsi); + } + } + break; + + /* + case LWS_CALLBACK_BROADCAST: + printf("LWS_CALLBACK_BROADCAST\n"); + n = libwebsocket_write(wsi, (unsigned char*)in, len, LWS_WRITE_TEXT); + if (n < 0) fprintf(stderr, "task write failed\n"); + break; + */ + + case LWS_CALLBACK_RECEIVE: + { + printf("LWS_CALLBACK_RECEIVE\n"); + + printf("%s\n", (char*)in); + std::string data; + data.append((char*)in, len); + + MessageList mlst = parse_msg(data); + printf("Handling %d incoming message\n", mlst.size()); + MessageList omsgs = handle_msg(mlst, wsi); + printf("Handling %d outgoing messages\n", omsgs.size()); + + MessageList::iterator omi = omsgs.begin(); + while(omi != omsgs.end()) { + std::string msg; + msg.append((size_t)LWS_SEND_BUFFER_PRE_PADDING, ' '); + msg.append(msg_tostring(*omi)); + msg.append((size_t)LWS_SEND_BUFFER_POST_PADDING, ' '); + + printf("%d nodes affected by command\n", omi->nodes.size()); + + std::set clients = + connection_handler.observerlist(omi->nodes); + printf("Writing message to %d clients\n", clients.size()); + + std::set::iterator ci = clients.begin(); + while(ci != clients.end()) { + clientid_t clientid = *ci; + printf("\tAdding data to %p's queue\n", clientid); + msgqueue[*ci].push(msg); + ci++; + } + + omi++; + } + + libwebsocket_callback_on_writable_all_protocol( + libwebsockets_get_protocol(wsi)); + } + break; + + default: + break; + } + + return 0; +} -- cgit v1.2.3