diff --git a/ChangeLog.txt b/ChangeLog.txt index c1756487..87bd6116 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -6,6 +6,8 @@ Broker: Closes #1934. - Fix DH group not being set for TLS connections, which meant ciphers using DHE couldn't be used. Closes #1925. Closes #1476. +- Fix websockets listeners not causing the main loop not to wake up. + Closes #1936. Client library: - Fix DH group not being set for TLS connections, which meant ciphers using diff --git a/src/mosquitto.c b/src/mosquitto.c index 5431ca16..eef66c44 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -58,6 +58,10 @@ Contributors: struct mosquitto_db db; +static struct mosquitto__listener_sock *listensock = NULL; +static int listensock_count = 0; +static int listensock_index = 0; + bool flag_reload = false; #ifdef WITH_PERSISTENCE bool flag_db_backup = false; @@ -202,7 +206,7 @@ void listeners__reload_all_certificates(void) } -int listeners__start_single_mqtt(struct mosquitto__listener_sock **listensock, int *listensock_count, int *listensock_index, struct mosquitto__listener *listener) +int listeners__start_single_mqtt(struct mosquitto__listener *listener) { int i; struct mosquitto__listener_sock *listensock_new; @@ -210,29 +214,66 @@ int listeners__start_single_mqtt(struct mosquitto__listener_sock **listensock, i if(net__socket_listen(listener)){ return 1; } - (*listensock_count) += listener->sock_count; - listensock_new = mosquitto__realloc(*listensock, sizeof(struct mosquitto__listener_sock)*(size_t)(*listensock_count)); + listensock_count += listener->sock_count; + listensock_new = mosquitto__realloc(listensock, sizeof(struct mosquitto__listener_sock)*(size_t)listensock_count); if(!listensock_new){ return 1; } - *listensock = listensock_new; + listensock = listensock_new; for(i=0; isock_count; i++){ if(listener->socks[i] == INVALID_SOCKET){ return 1; } - (*listensock)[*listensock_index].sock = listener->socks[i]; - (*listensock)[*listensock_index].listener = listener; + listensock[listensock_index].sock = listener->socks[i]; + listensock[listensock_index].listener = listener; #ifdef WITH_EPOLL - (*listensock)[*listensock_index].ident = id_listener; + listensock[listensock_index].ident = id_listener; #endif - (*listensock_index)++; + listensock_index++; } return MOSQ_ERR_SUCCESS; } -int listeners__add_local(struct mosquitto__listener_sock **listensock, int *listensock_count, int *listensock_index, const char *host, uint16_t port) +#ifdef WITH_WEBSOCKETS +void listeners__add_websockets(struct lws_context *ws_context, int fd) +{ + int i; + struct mosquitto__listener *listener = NULL; + struct mosquitto__listener_sock *listensock_new; + + /* Don't add more listeners after we've started the main loop */ + if(run || ws_context == NULL) return; + + /* Find context */ + for(i=0; ilistener_count; i++){ + if(db.config->listeners[i].ws_in_init){ + listener = &db.config->listeners[i]; + break; + } + } + if(listener == NULL){ + return; + } + + listensock_count++; + listensock_new = mosquitto__realloc(listensock, sizeof(struct mosquitto__listener_sock)*(size_t)listensock_count); + if(!listensock_new){ + return; + } + listensock = listensock_new; + + listensock[listensock_index].sock = fd; + listensock[listensock_index].listener = listener; +#ifdef WITH_EPOLL + listensock[listensock_index].ident = id_listener_ws; +#endif + listensock_index++; +} +#endif + +int listeners__add_local(const char *host, uint16_t port) { struct mosquitto__listener *listeners; listeners = db.config->listeners; @@ -244,7 +285,7 @@ int listeners__add_local(struct mosquitto__listener_sock **listensock, int *list if(listeners[db.config->listener_count].host == NULL){ return MOSQ_ERR_NOMEM; } - if(listeners__start_single_mqtt(listensock, listensock_count, listensock_index, &listeners[db.config->listener_count])){ + if(listeners__start_single_mqtt(&listeners[db.config->listener_count])){ mosquitto__free(listeners[db.config->listener_count].host); listeners[db.config->listener_count].host = NULL; return MOSQ_ERR_UNKNOWN; @@ -253,11 +294,10 @@ int listeners__add_local(struct mosquitto__listener_sock **listensock, int *list return MOSQ_ERR_SUCCESS; } -int listeners__start_local_only(struct mosquitto__listener_sock **listensock, int *listensock_count) +int listeners__start_local_only(void) { /* Attempt to open listeners bound to 127.0.0.1 and ::1 only */ int i; - int listensock_index = 0; int rc; struct mosquitto__listener *listeners; @@ -272,15 +312,15 @@ int listeners__start_local_only(struct mosquitto__listener_sock **listensock, in log__printf(NULL, MOSQ_LOG_WARNING, "Starting in local only mode. Connections will only be possible from clients running on this machine."); log__printf(NULL, MOSQ_LOG_WARNING, "Create a configuration file which defines a listener to allow remote access."); if(db.config->cmd_port_count == 0){ - rc = listeners__add_local(listensock, listensock_count, &listensock_index, "127.0.0.1", 1883); + rc = listeners__add_local("127.0.0.1", 1883); if(rc == MOSQ_ERR_NOMEM) return MOSQ_ERR_NOMEM; - rc = listeners__add_local(listensock, listensock_count, &listensock_index, "::1", 1883); + rc = listeners__add_local("::1", 1883); if(rc == MOSQ_ERR_NOMEM) return MOSQ_ERR_NOMEM; }else{ for(i=0; icmd_port_count; i++){ - rc = listeners__add_local(listensock, listensock_count, &listensock_index, "127.0.0.1", db.config->cmd_port[i]); + rc = listeners__add_local("127.0.0.1", db.config->cmd_port[i]); if(rc == MOSQ_ERR_NOMEM) return MOSQ_ERR_NOMEM; - rc = listeners__add_local(listensock, listensock_count, &listensock_index, "::1", db.config->cmd_port[i]); + rc = listeners__add_local("::1", db.config->cmd_port[i]); if(rc == MOSQ_ERR_NOMEM) return MOSQ_ERR_NOMEM; } } @@ -293,16 +333,14 @@ int listeners__start_local_only(struct mosquitto__listener_sock **listensock, in } -int listeners__start(struct mosquitto__listener_sock **listensock, int *listensock_count) +int listeners__start(void) { int i; - int listensock_index = 0; - listensock_index = 0; - (*listensock_count) = 0; + listensock_count = 0; if(db.config->listener_count == 0){ - if(listeners__start_local_only(listensock, listensock_count)){ + if(listeners__start_local_only()){ db__close(); if(db.config->pid_file){ (void)remove(db.config->pid_file); @@ -314,7 +352,7 @@ int listeners__start(struct mosquitto__listener_sock **listensock, int *listenso for(i=0; ilistener_count; i++){ if(db.config->listeners[i].protocol == mp_mqtt){ - if(listeners__start_single_mqtt(listensock, listensock_count, &listensock_index, &db.config->listeners[i])){ + if(listeners__start_single_mqtt(&db.config->listeners[i])){ db__close(); if(db.config->pid_file){ (void)remove(db.config->pid_file); @@ -323,7 +361,7 @@ int listeners__start(struct mosquitto__listener_sock **listensock, int *listenso } }else if(db.config->listeners[i].protocol == mp_websockets){ #ifdef WITH_WEBSOCKETS - db.config->listeners[i].ws_context = mosq_websockets_init(&db.config->listeners[i], db.config); + mosq_websockets_init(&db.config->listeners[i], db.config); if(!db.config->listeners[i].ws_context){ log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to create websockets listener on port %d.", db.config->listeners[i].port); return 1; @@ -331,7 +369,7 @@ int listeners__start(struct mosquitto__listener_sock **listensock, int *listenso #endif } } - if((*listensock) == NULL){ + if(listensock == NULL){ log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to start any listening sockets, exiting."); return 1; } @@ -339,7 +377,7 @@ int listeners__start(struct mosquitto__listener_sock **listensock, int *listenso } -void listeners__stop(struct mosquitto__listener_sock *listensock, int listensock_count) +void listeners__stop(void) { int i; @@ -404,8 +442,6 @@ int pid__write(void) int main(int argc, char *argv[]) { - struct mosquitto__listener_sock *listensock = NULL; - int listensock_count = 0; struct mosquitto__config config; #ifdef WITH_BRIDGE int i; @@ -510,7 +546,7 @@ int main(int argc, char *argv[]) sys_tree__init(); #endif - if(listeners__start(&listensock, &listensock_count)) return 1; + if(listeners__start()) return 1; signal__setup(); @@ -541,7 +577,7 @@ int main(int argc, char *argv[]) #endif session_expiry__remove_all(); - listeners__stop(listensock, listensock_count); + listeners__stop(); HASH_ITER(hh_id, db.contexts_by_id, ctxt, ctxt_tmp){ #ifdef WITH_WEBSOCKETS diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index 0a64f203..394c9cdf 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -186,6 +186,7 @@ enum struct_ident{ id_invalid = 0, id_listener = 1, id_client = 2, + id_listener_ws = 3, }; #endif @@ -224,6 +225,7 @@ struct mosquitto__listener { #endif #ifdef WITH_WEBSOCKETS struct lws_context *ws_context; + bool ws_in_init; char *http_dir; struct lws_protocols *ws_protocol; #endif @@ -557,6 +559,7 @@ struct mosquitto__bridge{ #ifdef WITH_WEBSOCKETS struct libws_mqtt_hack { char *http_dir; + struct mosquitto__listener *listener; }; struct libws_mqtt_data { @@ -743,6 +746,9 @@ int mux__cleanup(void); * ============================================================ */ void listener__set_defaults(struct mosquitto__listener *listener); void listeners__reload_all_certificates(void); +#ifdef WITH_WEBSOCKETS +void listeners__add_websockets(struct lws_context *ws_context, int fd); +#endif /* ============================================================ * Plugin related functions @@ -825,7 +831,7 @@ DWORD WINAPI SigThreadProc(void* data); * Websockets related functions * ============================================================ */ #ifdef WITH_WEBSOCKETS -struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, const struct mosquitto__config *conf); +void mosq_websockets_init(struct mosquitto__listener *listener, const struct mosquitto__config *conf); #endif void do_disconnect(struct mosquitto *context, int reason); diff --git a/src/mux_epoll.c b/src/mux_epoll.c index 23b23a16..2749845f 100644 --- a/src/mux_epoll.c +++ b/src/mux_epoll.c @@ -217,6 +217,11 @@ int mux_epoll__handle(void) mux__add_in(context); } } +#ifdef WITH_WEBSOCKETS + }else if(context->ident == id_listener_ws){ + /* Nothing needs to happen here, because we always call lws_service in the loop. + * The important point is we've been woken up for this listener. */ +#endif } } } diff --git a/src/mux_poll.c b/src/mux_poll.c index a31e9453..fb8f56d3 100644 --- a/src/mux_poll.c +++ b/src/mux_poll.c @@ -211,9 +211,17 @@ int mux_poll__handle(struct mosquitto__listener_sock *listensock, int listensock for(i=0; ipollfd_index = -1; - mux__add_in(context); +#ifdef WITH_WEBSOCKETS + if(listensock[i]->listener.ws_context){ + /* Nothing needs to happen here, because we always call lws_service in the loop. + * The important point is we've been woken up for this listener. */ + }else +#endif + { + while((context = net__socket_accept(&listensock[i])) != NULL){ + context->pollfd_index = -1; + mux__add_in(context); + } } } } diff --git a/src/websockets.c b/src/websockets.c index 53c8379a..597f59da 100644 --- a/src/websockets.c +++ b/src/websockets.c @@ -616,6 +616,11 @@ static int callback_http( }else{ mux__remove_out(mosq); } + }else{ + if(reason == LWS_CALLBACK_ADD_POLL_FD && (pollargs->events & POLLIN)){ + /* Assume this is a new listener */ + listeners__add_websockets(lws_get_context(wsi), pollargs->fd); + } } break; @@ -641,7 +646,7 @@ static void log_wrap(int level, const char *line) log__printf(NULL, MOSQ_LOG_WEBSOCKETS, "%s", l); } -struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, const struct mosquitto__config *conf) +void mosq_websockets_init(struct mosquitto__listener *listener, const struct mosquitto__config *conf) { struct lws_context_creation_info info; struct lws_protocols *p; @@ -655,7 +660,7 @@ struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, c p = mosquitto__calloc(protocol_count+1, sizeof(struct lws_protocols)); if(!p){ log__printf(NULL, MOSQ_LOG_ERR, "Out of memory."); - return NULL; + return; } for(i=0; protocols[i].name; i++){ p[i].name = protocols[i].name; @@ -694,7 +699,7 @@ struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, c if(!user){ mosquitto__free(p); log__printf(NULL, MOSQ_LOG_ERR, "Out of memory."); - return NULL; + return; } if(listener->http_dir){ @@ -707,9 +712,10 @@ struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, c mosquitto__free(user); mosquitto__free(p); log__printf(NULL, MOSQ_LOG_ERR, "Error: Unable to open http dir \"%s\".", listener->http_dir); - return NULL; + return; } } + user->listener = listener; info.user = user; info.pt_serv_buf_size = WS_SERV_BUF_SIZE; @@ -718,7 +724,9 @@ struct lws_context *mosq_websockets_init(struct mosquitto__listener *listener, c lws_set_log_level(conf->websockets_log_level, log_wrap); log__printf(NULL, MOSQ_LOG_INFO, "Opening websockets listen socket on port %d.", listener->port); - return lws_create_context(&info); + listener->ws_in_init = true; + listener->ws_context = lws_create_context(&info); + listener->ws_in_init = false; }