--- libaitmqtt/src/srvside.c 2012/01/30 08:33:37 1.1.2.3 +++ libaitmqtt/src/srvside.c 2012/06/29 13:54:48 1.3 @@ -1,21 +1,71 @@ -#include "global.h" +/************************************************************************* +* (C) 2011 AITNET ltd - Sofia/Bulgaria - +* by Michael Pounov +* +* $Author: misho $ +* $Id: srvside.c,v 1.3 2012/06/29 13:54:48 misho Exp $ +* +************************************************************************** +The ELWIX and AITNET software is distributed under the following +terms: +All of the documentation and software included in the ELWIX and AITNET +Releases is copyrighted by ELWIX - Sofia/Bulgaria +Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012 + by Michael Pounov . All rights reserved. +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions +are met: +1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. +2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. +3. All advertising materials mentioning features or use of this software + must display the following acknowledgement: +This product includes software developed by Michael Pounov +ELWIX - Embedded LightWeight unIX and its contributors. +4. Neither the name of AITNET nor the names of its contributors + may be used to endorse or promote products derived from this software + without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY AITNET AND CONTRIBUTORS ``AS IS'' AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF +SUCH DAMAGE. +*/ +#include "global.h" + + /* - * mqtt_srv_Init() Init MQTT server side support + * mqtt_srv_cliInit() Init MQTT server side support for clients * * @sock = Client socket from accept() * @buf = Message buffer + * @timeout = timeout + * @nb = Non block socket * return: NULL error or !=NULL allocated server handle */ mqtt_srv_t * -mqtt_srv_Init(int sock, mqtt_msg_t * __restrict buf) +mqtt_srv_cliInit(int sock, mqtt_msg_t * __restrict buf, u_short timeout, int nb) { mqtt_srv_t *srv = NULL; + struct timeval tv = { timeout, 0 }; + int n; if (!buf || !buf->msg_base || !buf->msg_len) return srv; + else + n = buf->msg_len; srv = malloc(sizeof(mqtt_srv_t)); if (!srv) { @@ -24,20 +74,28 @@ mqtt_srv_Init(int sock, mqtt_msg_t * __restrict buf) } else memset(srv, 0, sizeof(mqtt_srv_t)); + /* set options of client socket */ + setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &n, sizeof n); + setsockopt(sock, SOL_SOCKET, SO_RCVBUF, &n, sizeof n); + setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv); + setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv); + ioctl(sock, FIONBIO, nb); + srv->sock = sock; srv->buf = buf; + srv->timeout = timeout; return srv; } /* - * mqtt_srv_Fini() Finish MQTT server side support + * mqtt_srv_cliFini() Finish MQTT server side support for clients * * @psrv = Server handle * return: none */ void -mqtt_srv_Fini(mqtt_srv_t ** __restrict psrv) +mqtt_srv_cliFini(mqtt_srv_t ** __restrict psrv) { if (psrv && *psrv) { free(*psrv); @@ -46,31 +104,103 @@ mqtt_srv_Fini(mqtt_srv_t ** __restrict psrv) } /* - * mqtt_srv_Dispatch() MQTT server dispatcher + * mqtt_srv_cliDispatch() MQTT server dispatcher for clients * * @srv = Server handle + * @rlen = Ready bytes to parse * @arg = Argument pass to command callback * return: -1 error or 0 ok */ int -mqtt_srv_Dispatch(mqtt_srv_t * __restrict srv, void *arg) +mqtt_srv_cliDispatch(mqtt_srv_t * __restrict srv, int rlen, void *arg) { struct mqtthdr *hdr; if (!srv || !srv->buf->msg_base || !srv->buf->msg_len) return -1; - if (srv->buf->msg_len < sizeof(struct mqtthdr)) { - mqtt_SetErr(EINVAL, "Error:: message is too small ..."); + if (srv->buf->msg_len < sizeof(struct mqtthdr) || rlen < sizeof(struct mqtthdr)) { + mqtt_SetErr(EINVAL, "Message is too short ..."); return -1; } hdr = (struct mqtthdr*) srv->buf->msg_base; if (hdr->mqtt_msg.type > MQTT_TYPE_MAX) { - mqtt_SetErr(ENOSYS, "Error:: message type %d not implemented", hdr->mqtt_msg.type); + mqtt_SetErr(ENOSYS, "Message type %d not implemented", hdr->mqtt_msg.type); return -1; } if (srv->cmds[hdr->mqtt_msg.type]) - return srv->cmds[hdr->mqtt_msg.type]((void*) srv, arg); + return srv->cmds[hdr->mqtt_msg.type]((void*) srv, rlen, arg); - mqtt_SetErr(ENOTSUP, "Error:: Unsupported message type %d", hdr->mqtt_msg.type); + mqtt_SetErr(ENOTSUP, "Unsupported message type %d", hdr->mqtt_msg.type); return -1; +} + + +/* + * mqtt_srv_Create() - Create server socket + * + * @sa = Server bind address + * @salen = Server struct sockaddr size + * return: -1 error or >-1 server socket + */ +int +mqtt_srv_Create(struct sockaddr * __restrict sa, int salen) +{ + int s = -1, n = 1; + + if (!sa) + return -1; + + s = socket(sa->sa_family, SOCK_STREAM, 0); + if (s == -1) { + LOGERR; + return -1; + } + if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &n, sizeof n) == -1) { + LOGERR; + close(s); + return -1; + } + if (bind(s, sa, salen) == -1) { + LOGERR; + close(s); + return -1; + } + + return s; +} + +/* + * mqtt_srv_Destroy() - Close server socket + * + * @sock = Server socket + * return: -1 error or 0 ok + */ +int +mqtt_srv_Destroy(int sock) +{ + shutdown(sock, SHUT_RDWR); + return close(sock); +} + +/* + * mqtt_srv_Listen() - Listen server socket + * + * @sock = Server socket + * @maxconn = max number of pending connections + * @nb = Non block socket + * return: -1 error or 0 ok + */ +int +mqtt_srv_Listen(int sock, int maxconn, int nb) +{ + if (!maxconn) + maxconn = SOMAXCONN; + + if (listen(sock, maxconn) == -1) { + LOGERR; + return -1; + } else + ioctl(sock, FIONBIO, nb); + + return 0; }