diff --git a/MQTTSNGateway/src/MQTTConnection.h b/MQTTSNGateway/src/MQTTConnection.h new file mode 100644 index 0000000..e2df956 --- /dev/null +++ b/MQTTSNGateway/src/MQTTConnection.h @@ -0,0 +1,559 @@ +/******************************************************************************* + * Copyright (c) 2014, 2016 IBM Corp. + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * Contributors: + * Ian Craggs - initial API and implementation and/or initial documentation + *******************************************************************************/ + +#if !defined(MQTTCONNECTION_H) +#define MQTTCONNECTION_H + +#include "FP.h" +#include "MQTTPacket.h" +#include "stdio.h" +#include "MQTTLogging.h" + +#if !defined(MQTTCLIENT_QOS1) + #define MQTTCLIENT_QOS1 1 +#endif +#if !defined(MQTTCLIENT_QOS2) + #define MQTTCLIENT_QOS2 0 +#endif + +namespace MQTT +{ + + +enum QoS { QOS0, QOS1, QOS2 }; + +// all failure return codes must be negative +enum returnCode { BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 }; + + +struct Message +{ + enum QoS qos; + bool retained; + bool dup; + unsigned short id; + void *payload; + size_t payloadlen; +}; + + +class PacketId +{ +public: + PacketId() + { + next = 0; + } + + int getNext() + { + return next = (next == MAX_PACKET_ID) ? 1 : ++next; + } + +private: + static const int MAX_PACKET_ID = 65535; + int next; +}; + + +/** + * @class Client + * @brief blocking, non-threaded MQTT client API + * + * This version of the API blocks on all method calls, until they are complete. This means that only one + * MQTT request can be in process at any one time. + * @param Network a network class which supports send, receive + * @param Timer a timer class with the methods: + */ +template +class Connection +{ + +public: + + /** Construct the client + * @param network - pointer to an instance of the Network class - must be connected to the endpoint + * before calling MQTT connect + * @param limits an instance of the Limit class - to alter limits as required + */ + Connection(unsigned int command_timeout_ms = 30000); + + + static void run(void const* arg); + + + /** MQTT Disconnect - send an MQTT disconnect packet, and clean up any state + * @return success code - + */ + int disconnect(); + + /** Is the client connected? + * @return flag - is the client connected or not? + */ + bool isConnected() + { + return isconnected; + } + + int sendPacket(int length, Timer& timer); + unsigned char sendbuf[MAX_MQTT_PACKET_SIZE]; + +private: + + int connect(MQTTPacket_connectData&); + int connect(); + + void cleanSession(); + int cycle(Timer& timer); + int waitfor(int packet_type, Timer& timer); + int keepalive(); + int publish(int len, Timer& timer, enum QoS qos); + + int decodePacket(int* value, int timeout); + int readPacket(Timer& timer); + + Network ipstack; + unsigned long command_timeout_ms; + unsigned char readbuf[MAX_MQTT_PACKET_SIZE]; + + Timer last_sent, last_received; + unsigned int keepAliveInterval; + bool ping_outstanding; + bool cleansession; + + PacketId packetid; + + bool isconnected; + +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + unsigned char pubbuf[MAX_MQTT_PACKET_SIZE]; // store the last publish for sending on reconnect + int inflightLen; + unsigned short inflightMsgid; + enum QoS inflightQoS; +#endif + +#if MQTTCLIENT_QOS2 + bool pubrel; + #if !defined(MAX_INCOMING_QOS2_MESSAGES) + #define MAX_INCOMING_QOS2_MESSAGES 10 + #endif + unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES]; + bool isQoS2msgidFree(unsigned short id); + bool useQoS2msgid(unsigned short id); + void freeQoS2msgid(unsigned short id); +#endif + +}; + +} + + +template +void MQTT::Connection::run(void const* arg) +{ + + + + +} + + +template +void MQTT::Connection::cleanSession() +{ + ping_outstanding = false; + isconnected = false; + +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + inflightMsgid = 0; + inflightQoS = QOS0; +#endif + +#if MQTTCLIENT_QOS2 + pubrel = false; + for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) + incomingQoS2messages[i] = 0; +#endif +} + + +template +MQTT::Connection::Connection(unsigned int command_timeout_ms) : packetid() +{ + last_sent = Timer(); + last_received = Timer(); + this->command_timeout_ms = command_timeout_ms; + cleanSession(); +} + + +#if MQTTCLIENT_QOS2 +template +bool MQTT::Connection::isQoS2msgidFree(unsigned short id) +{ + for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) + { + if (incomingQoS2messages[i] == id) + return false; + } + return true; +} + + +template +bool MQTT::Connection::useQoS2msgid(unsigned short id) +{ + for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) + { + if (incomingQoS2messages[i] == 0) + { + incomingQoS2messages[i] = id; + return true; + } + } + return false; +} + + +template +void MQTT::Connection::freeQoS2msgid(unsigned short id) +{ + for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i) + { + if (incomingQoS2messages[i] == id) + { + incomingQoS2messages[i] = 0; + return; + } + } +} +#endif + + +template +int MQTT::Connection::sendPacket(int length, Timer& timer) +{ + int rc = FAILURE, + sent = 0; + + while (sent < length && !timer.expired()) + { + rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms()); + if (rc < 0) // there was an error writing the data + break; + sent += rc; + } + if (sent == length) + { + if (this->keepAliveInterval > 0) + last_sent.countdown(this->keepAliveInterval); // record the fact that we have successfully sent the packet + rc = SUCCESS; + } + else + rc = FAILURE; + +#if defined(MQTT_DEBUG) + char printbuf[150]; + DEBUG("Rc %d from sending packet %s\n", rc, MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length)); +#endif + return rc; +} + + +template +int MQTT::Connection::decodePacket(int* value, int timeout) +{ + unsigned char c; + int multiplier = 1; + int len = 0; + const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4; + + *value = 0; + do + { + int rc = MQTTPACKET_READ_ERROR; + + if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES) + { + rc = MQTTPACKET_READ_ERROR; /* bad data */ + goto exit; + } + rc = ipstack.read(&c, 1, timeout); + if (rc != 1) + goto exit; + *value += (c & 127) * multiplier; + multiplier *= 128; + } while ((c & 128) != 0); +exit: + return len; +} + + +/** + * If any read fails in this method, then we should disconnect from the network, as on reconnect + * the packets can be retried. + * @param timeout the max time to wait for the packet read to complete, in milliseconds + * @return the MQTT packet type, or -1 if none + */ +template +int MQTT::Connection::readPacket(Timer& timer) +{ + int rc = FAILURE; + MQTTHeader header = {0}; + int len = 0; + int rem_len = 0; + + /* 1. read the header byte. This has the packet type in it */ + if (ipstack.read(readbuf, 1, timer.left_ms()) != 1) + goto exit; + + len = 1; + /* 2. read the remaining length. This is variable in itself */ + decodePacket(&rem_len, timer.left_ms()); + len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */ + + if (rem_len > (MAX_MQTT_PACKET_SIZE - len)) + { + rc = BUFFER_OVERFLOW; + goto exit; + } + + /* 3. read the rest of the buffer using a callback to supply the rest of the data */ + if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len)) + goto exit; + + header.byte = readbuf[0]; + rc = header.bits.type; + if (this->keepAliveInterval > 0) + last_received.countdown(this->keepAliveInterval); // record the fact that we have successfully received a packet +exit: + +#if defined(MQTT_DEBUG) + if (rc >= 0) + { + char printbuf[50]; + DEBUG("Rc %d from receiving packet %s\n", rc, MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len)); + } +#endif + return rc; +} + + +template +int MQTT::Connection::cycle(Timer& timer) +{ + /* get one piece of work off the wire and one pass through */ + + // read the socket, see what work is due + int packet_type = readPacket(timer); + + int len = 0, + rc = SUCCESS; + + switch (packet_type) + { + case FAILURE: + case BUFFER_OVERFLOW: + rc = packet_type; + break; + case CONNACK: + case PUBACK: + case SUBACK: + break; + case PUBLISH: + { + MQTTString topicName = MQTTString_initializer; + Message msg; + int intQoS; + if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName, + (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1) + goto exit; + msg.qos = (enum QoS)intQoS; +#if MQTTCLIENT_QOS2 + if (msg.qos != QOS2) +#endif + ; //deliverMessage(topicName, msg); +#if MQTTCLIENT_QOS2 + else if (isQoS2msgidFree(msg.id)) + { + if (useQoS2msgid(msg.id)) + ; //deliverMessage(topicName, msg); + else + WARN("Maximum number of incoming QoS2 messages exceeded"); + } +#endif +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + if (msg.qos != QOS0) + { + if (msg.qos == QOS1) + len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBACK, 0, msg.id); + else if (msg.qos == QOS2) + len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREC, 0, msg.id); + if (len <= 0) + rc = FAILURE; + else + rc = sendPacket(len, timer); + if (rc == FAILURE) + goto exit; // there was a problem + } + break; +#endif + } +#if MQTTCLIENT_QOS2 + case PUBREC: + case PUBREL: + unsigned short mypacketid; + unsigned char dup, type; + if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1) + rc = FAILURE; + else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, + (packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid)) <= 0) + rc = FAILURE; + else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet + rc = FAILURE; // there was a problem + if (rc == FAILURE) + goto exit; // there was a problem + if (packet_type == PUBREL) + freeQoS2msgid(mypacketid); + break; + + case PUBCOMP: + break; +#endif + case PINGRESP: + ping_outstanding = false; + break; + } + keepalive(); +exit: + if (rc == SUCCESS) + rc = packet_type; + return rc; +} + + +template +int MQTT::Connection::keepalive() +{ + int rc = FAILURE; + + if (keepAliveInterval == 0) + { + rc = SUCCESS; + goto exit; + } + + if (last_sent.expired() || last_received.expired()) + { + if (!ping_outstanding) + { + Timer timer(1000); + int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE); + if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet + ping_outstanding = true; + } + } + +exit: + return rc; +} + + +template +int MQTT::Connection::connect(MQTTPacket_connectData& options) +{ + Timer connect_timer(command_timeout_ms); + int rc = FAILURE; + int len = 0; + + if (isconnected) // don't send connect packet again if we are already connected + goto exit; + + this->keepAliveInterval = options.keepAliveInterval; + this->cleansession = options.cleansession; + if ((len = MQTTSerialize_connect(sendbuf, MAX_MQTT_PACKET_SIZE, &options)) <= 0) + goto exit; + if ((rc = sendPacket(len, connect_timer)) != SUCCESS) // send the connect packet + goto exit; // there was a problem + + if (this->keepAliveInterval > 0) + last_received.countdown(this->keepAliveInterval); + // this will be a blocking call, wait for the connack + if (waitfor(CONNACK, connect_timer) == CONNACK) + { + unsigned char connack_rc = 255; + bool sessionPresent = false; + if (MQTTDeserialize_connack((unsigned char*)&sessionPresent, &connack_rc, readbuf, MAX_MQTT_PACKET_SIZE) == 1) + rc = connack_rc; + else + rc = FAILURE; + } + else + rc = FAILURE; + +#if MQTTCLIENT_QOS2 + // resend any inflight publish + if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel) + { + if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0) + rc = FAILURE; + else + rc = publish(len, connect_timer, inflightQoS); + } + else +#endif +#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2 + if (inflightMsgid > 0) + { + memcpy(sendbuf, pubbuf, MAX_MQTT_PACKET_SIZE); + rc = publish(inflightLen, connect_timer, inflightQoS); + } +#endif + +exit: + if (rc == SUCCESS) + isconnected = true; + return rc; +} + + +template +int MQTT::Connection::connect() +{ + MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer; + return connect(default_options); +} + + + +template +int MQTT::Connection::disconnect() +{ + int rc = FAILURE; + Timer timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete + int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE); + if (len > 0) + rc = sendPacket(len, timer); // send the disconnect packet + + if (cleansession) + cleanSession(); + else + isconnected = false; + return rc; +} + + +#endif diff --git a/MQTTSNGateway/src/MQTTSNGateway.h b/MQTTSNGateway/src/MQTTSNGateway.h new file mode 100644 index 0000000..94ffa59 --- /dev/null +++ b/MQTTSNGateway/src/MQTTSNGateway.h @@ -0,0 +1,296 @@ +/******************************************************************************* + * Copyright (c) 2014, 2016 IBM Corp. + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * Contributors: + * Ian Craggs - initial API and implementation and/or initial documentation + *******************************************************************************/ + +#if !defined(MQTTSNGATEWAY_H) +#define MQTTSNGATEWAY_H + +#include "MQTTSNPacket.h" +#include "MQTTConnection.h" + +namespace MQTTSN +{ + +#define MAX_PACKET_SIZE 256 +#define MQTTSNGATEWAY_QOS2 0 +#define MQTTSNGATEWAY_QOS1 0 +#define MAX_MQTT_CONNECTIONS 5 + +struct Parms +{ + char* hostname; + int port; + char* username; + char* password; +}; + +enum QoS { QOS0, QOS1, QOS2 }; + +enum Modes { AGGREGATING, TRANSPARENT }; + +// all failure return codes must be negative +enum returnCode { MAX_SUBSCRIPTIONS_EXCEEDED = -3, BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 }; + +struct Message +{ + enum QoS qos; + bool retained; + bool dup; + unsigned short id; + void *payload; + size_t payloadlen; +}; + + +template +class Gateway +{ +public: + + Gateway(UDPNetwork&, unsigned int command_timeout_ms = 30000); + + void run(void const* arg); + +private: + + int cycle(Timer& timer); + int sendPacket(int length, Timer& timer); + int readPacket(Timer& timer); + + UDPNetwork& udpstack; // Not restricted to UDP - for want of a better name + + unsigned char sendbuf[MAX_PACKET_SIZE]; + unsigned char readbuf[MAX_PACKET_SIZE]; + + enum Modes mode; + + MQTT::Connection connections[MAX_MQTT_CONNECTIONS]; + +}; + + +} // end namespace + + +template +MQTTSN::Gateway::Gateway(UDPNetwork& network, unsigned int command_timeout_ms) : udpstack(network) +{ + mode = AGGREGATING; +} + + +template +void MQTTSN::Gateway::run(void const* arg) +{ + Timer timer; + + printf("Gateway run 0\n"); + if (mode == AGGREGATING) + { + // set up connection 0 information + Thread mythread(&connections[0].run); + } + + printf("Gateway run 1\n"); + while (true) + { + printf("Gateway cycle\n"); + cycle(timer); + } + +} + + +template +int MQTTSN::Gateway::cycle(Timer& timer) +{ + /* get one piece of work off the wire and one pass through */ + + // read the socket, see what work is due + unsigned short packet_type = readPacket(timer); + + printf("read packet\n"); + + int len = 0; + int rc = SUCCESS; + + switch (packet_type) + { + case MQTTSN_CONNECT: + { + MQTTSNPacket_connectData data = MQTTSNPacket_connectData_initializer; + MQTTSNDeserialize_connect(&data, readbuf, MAX_PACKET_SIZE); + + if (mode == TRANSPARENT) + { + //start a new MQTT connection at this point + } + + len = MQTTSNSerialize_connack(sendbuf, MAX_PACKET_SIZE, 0); + if (len <= 0) + rc = FAILURE; + else + rc = sendPacket(len, timer); + break; + } + + case MQTTSN_REGISTER: + { + unsigned short topicid, packetid; + MQTTSNString topicName; + unsigned char reg_rc = MQTTSN_RC_ACCEPTED; + if (MQTTSNDeserialize_register(&topicid, &packetid, &topicName, readbuf, MAX_PACKET_SIZE) != 1) + goto exit; + + // store topic registration info + + len = MQTTSNSerialize_regack(connections[0].sendbuf, MAX_PACKET_SIZE, topicid, packetid, reg_rc); + if (len <= 0) + rc = FAILURE; + else + rc = connections[0].sendPacket(len, timer); + + break; + } + + case MQTTSN_SUBSCRIBE: + { + unsigned char dup; + int qos; + unsigned short packetid; + MQTTSN_topicid topicFilter; + MQTTSNDeserialize_subscribe(&dup, &qos, &packetid, &topicFilter, readbuf, MAX_PACKET_SIZE); + MQTTString topic = MQTTString_initializer; + + if (topicFilter.type == MQTTSN_TOPIC_TYPE_NORMAL) + { + topic.lenstring.len = topicFilter.data.long_.len; + topic.lenstring.data = topicFilter.data.long_.name; + } + + len = MQTTSerialize_subscribe(connections[0].sendbuf, MAX_PACKET_SIZE, 0, packetid, 1, &topic, (int*)&qos); + if (len <= 0) + goto exit; + if ((rc = connections[0].sendPacket(len, timer)) != SUCCESS) // send the subscribe packet + goto exit; // there was a problem + + break; + } + + case MQTTSN_PUBLISH: + { + MQTTString topic = MQTTString_initializer; + MQTTSN_topicid topicid; + Message msg; + + if (MQTTSNDeserialize_publish((unsigned char*)&msg.dup, (int*)&msg.qos, (unsigned char*)&msg.retained, &msg.id, &topicid, + (unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_PACKET_SIZE) != 1) + goto exit; + + + len = MQTTSerialize_publish(connections[0].sendbuf, MAX_PACKET_SIZE, msg.dup, msg.qos, msg.retained, msg.id, topic, + (unsigned char*)msg.payload, msg.payloadlen); + if (len <= 0) + goto exit; + if ((rc = connections[0].sendPacket(len, timer)) != SUCCESS) // send the subscribe packet + goto exit; // there was a problem + + } + + + case MQTTSN_PINGRESP: + //ping_outstanding = false; + break; + } + //keepalive(); +exit: + if (rc == SUCCESS) + rc = packet_type; + return rc; +} + + + +template +int MQTTSN::Gateway::sendPacket(int length, Timer& timer) +{ + int rc = FAILURE, + sent = 0; + + do + { + sent = udpstack.write(sendbuf, length, timer.left_ms()); + printf("sendPacket, rc %d from write of %d bytes\n", sent, length); + if (sent < 0) // there was an error writing the data + break; + } + while (sent != length && !timer.expired()); + + if (sent == length) + { + //if (this->duration > 0) + // last_sent.countdown(this->duration); // record the fact that we have successfully sent the packet + rc = SUCCESS; + } + else + rc = FAILURE; + +#if defined(MQTT_DEBUG) + char printbuf[50]; + DEBUG("Rc %d from sending packet %s\n", rc, MQTTSNPacket_toString(printbuf, sizeof(printbuf), sendbuf, length)); +#endif + return rc; +} + + +/** + * If any read fails in this method, then we should disconnect from the network, as on reconnect + * the packets can be retried. + * @param timeout the max time to wait for the packet read to complete, in milliseconds + * @return the MQTT packet type, or -1 if none + */ +template +int MQTTSN::Gateway::readPacket(Timer& timer) +{ + int rc = FAILURE; + int len = 0; // the length of the whole packet including length field + int lenlen = 0; + int datalen = 0; + + #define MQTTSN_MIN_PACKET_LENGTH 3 + // 1. read the packet, datagram style + if ((len = udpstack.read(readbuf, MAX_PACKET_SIZE, timer.left_ms())) < MQTTSN_MIN_PACKET_LENGTH) + goto exit; + + // 2. read the length. This is variable in itself + lenlen = MQTTSNPacket_decode(readbuf, len, &datalen); + if (datalen != len) + goto exit; // there was an error + + rc = readbuf[lenlen]; + //if (this->duration > 0) + // last_received.countdown(this->duration); // record the fact that we have successfully received a packet +exit: + +#if defined(MQTT_DEBUG) + char printbuf[50]; + DEBUG("Rc %d from receiving packet %s\n", rc, MQTTSNPacket_toString(printbuf, sizeof(printbuf), readbuf, len)); +#endif + return rc; +} + + + +#endif diff --git a/MQTTSNGateway/src/linux/Threading.h b/MQTTSNGateway/src/linux/Threading.h new file mode 100644 index 0000000..016421e --- /dev/null +++ b/MQTTSNGateway/src/linux/Threading.h @@ -0,0 +1,36 @@ + + +extern "C" +{ + #include "Thread.h" +} + + +class Thread +{ + +/* +Thread(void (*fn)(void const *argument), void *argument) +{ + Thread_start(fn, arg); +}*/ + +public: + + Thread(void (*fn)(void const *argument)) + { + const void* arg = NULL; + + //Thread_start((void (*)(void *))fn, arg); + } + +}; + + + + +class Mutex +{ + + +}; diff --git a/MQTTSNGateway/src/linux/build.sh b/MQTTSNGateway/src/linux/build.sh new file mode 100644 index 0000000..e597508 --- /dev/null +++ b/MQTTSNGateway/src/linux/build.sh @@ -0,0 +1,2 @@ +g++ main.cpp -I MQTTPacket/src -I MQTTSNPacket/src -I linux MQTTPacket/src/*.c MQTTSNPacket/src/*.c + diff --git a/MQTTSNGateway/src/linux/linux.cpp b/MQTTSNGateway/src/linux/linux.cpp new file mode 100644 index 0000000..222acda --- /dev/null +++ b/MQTTSNGateway/src/linux/linux.cpp @@ -0,0 +1,224 @@ +/******************************************************************************* + * Copyright (c) 2014, 2015 IBM Corp. + * + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. + * + * Contributors: + * Ian Craggs - initial API and implementation and/or initial documentation + *******************************************************************************/ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + + +class IPStack +{ +public: + IPStack() + { + + } + + int Socket_error(const char* aString) + { + int rc = 0; + //if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) + //{ + if (strcmp(aString, "shutdown") != 0 || (errno != ENOTCONN && errno != ECONNRESET)) + { + if (errno != EINTR && errno != EAGAIN && errno != EINPROGRESS && errno != EWOULDBLOCK) + printf("Socket error %s in %s for socket %d\n", strerror(errno), aString, mysock); + rc = errno; + } + //} + return errno; + } + + int connect(const char* hostname, int port) + { + int type = SOCK_STREAM; + struct sockaddr_in address; + int rc = -1; + sa_family_t family = AF_INET; + struct addrinfo *result = NULL; + struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL}; + + if ((rc = getaddrinfo(hostname, NULL, &hints, &result)) == 0) + { + struct addrinfo* res = result; + + /* prefer ip4 addresses */ + while (res) + { + if (res->ai_family == AF_INET) + { + result = res; + break; + } + res = res->ai_next; + } + + if (result->ai_family == AF_INET) + { + address.sin_port = htons(port); + address.sin_family = family = AF_INET; + address.sin_addr = ((struct sockaddr_in*)(result->ai_addr))->sin_addr; + } + else + rc = -1; + + freeaddrinfo(result); + } + + if (rc == 0) + { + mysock = socket(family, type, 0); + if (mysock != -1) + { + int opt = 1; + + //if (setsockopt(mysock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0) + // printf("Could not set SO_NOSIGPIPE for socket %d", mysock); + + rc = ::connect(mysock, (struct sockaddr*)&address, sizeof(address)); + } + } + + return rc; + } + + int read(unsigned char* buffer, int len, int timeout_ms) + { + struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000}; + if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0)) + { + interval.tv_sec = 0; + interval.tv_usec = 100; + } + + setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval)); + + int bytes = 0; + while (bytes < len) + { + int rc = ::recv(mysock, &buffer[bytes], (size_t)(len - bytes), 0); + if (rc == -1) + { + if (Socket_error("read") != 0) + { + bytes = -1; + break; + } + } + else + bytes += rc; + } + return bytes; + } + + int write(unsigned char* buffer, int len, int timeout) + { + struct timeval tv; + + tv.tv_sec = 0; /* 30 Secs Timeout */ + tv.tv_usec = timeout * 1000; // Not init'ing this can cause strange errors + + setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv,sizeof(struct timeval)); + int rc = ::write(mysock, buffer, len); + //printf("write rc %d\n", rc); + return rc; + } + + int disconnect() + { + return ::close(mysock); + } + +private: + + int mysock; + +}; + + +class Countdown +{ +public: + Countdown() + { + + } + + Countdown(int ms) + { + countdown_ms(ms); + } + + + bool expired() + { + struct timeval now, res; + gettimeofday(&now, NULL); + timersub(&end_time, &now, &res); + //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); + //if (res.tv_sec > 0 || res.tv_usec > 0) + // printf("expired %d %d\n", res.tv_sec, res.tv_usec); + return res.tv_sec < 0 || (res.tv_sec == 0 && res.tv_usec <= 0); + } + + + void countdown_ms(int ms) + { + struct timeval now; + gettimeofday(&now, NULL); + struct timeval interval = {ms / 1000, (ms % 1000) * 1000}; + //printf("interval %d %d\n", interval.tv_sec, interval.tv_usec); + timeradd(&now, &interval, &end_time); + } + + + void countdown(int seconds) + { + struct timeval now; + gettimeofday(&now, NULL); + struct timeval interval = {seconds, 0}; + timeradd(&now, &interval, &end_time); + } + + + int left_ms() + { + struct timeval now, res; + gettimeofday(&now, NULL); + timersub(&end_time, &now, &res); + //printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000); + return (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000; + } + +private: + + struct timeval end_time; +}; + diff --git a/MQTTSNGateway/src/linux/main.cpp b/MQTTSNGateway/src/linux/main.cpp new file mode 100644 index 0000000..3505782 --- /dev/null +++ b/MQTTSNGateway/src/linux/main.cpp @@ -0,0 +1,24 @@ + +#include "Threading.h" + +//#include "MQTTSNUDP.h" +//#include "MQTTEthernet.h" +#include "linux.cpp" +#include "MQTTSNGateway.h" + +MQTTSN::Parms parms(); + + +int main() +{ + // set up MQTT-SN network listening + UDPStack net; + net.listen(1884); + + MQTTSN::Gateway gateway = + MQTTSN::Gateway(net); + + gateway.run(NULL); + + return 0; +} diff --git a/MQTTSNGateway/src/main.cpp b/MQTTSNGateway/src/main.cpp new file mode 100644 index 0000000..ae9c9fb --- /dev/null +++ b/MQTTSNGateway/src/main.cpp @@ -0,0 +1,33 @@ +/* + * main.cpp + * + * Created on: Jan 5, 2016 + * Author: icraggs + */ + + +void read_config() +{ + +} + + + +int main(int argc, char* argv[]) +{ + + + // 1. listen for MQTT-SN connections + + // 2. connect to MQTT broker + + // 3. listen for MQTT-SN packets, and act on them + // 4. listen for MQTT packets too, and act on them + + + + + + return 0; +} +