MQTTSocket.h

00001 #if !defined(MQTTSOCKET_H)
00002 #define MQTTSOCKET_H
00003 
00004 #include "MQTTmbed.h"
00005 #include <EthernetInterface.h>
00006 #include <Timer.h>
00007 
00008 class MQTTSocket
00009 {
00010 public:
00011     MQTTSocket(EthernetInterface *anet)
00012     {
00013         net = anet;
00014         open = false;
00015     }
00016     
00017     int connect(char* hostname, int port, int timeout=1000)
00018     {
00019         if (open)
00020             disconnect();
00021         nsapi_error_t rc = mysock.open(net);
00022         open = true;
00023         mysock.set_blocking(true);
00024         mysock.set_timeout((unsigned int)timeout);  
00025         SocketAddress socketAddr(hostname,port);
00026         rc = mysock.connect(socketAddr);
00027         mysock.set_blocking(false);  // blocking timeouts seem not to work
00028         return rc;
00029     }
00030 
00031     // common read/write routine, avoiding blocking timeouts
00032     int common(unsigned char* buffer, int len, int timeout, bool read)
00033     {
00034         timer.start();
00035         mysock.set_blocking(false); // blocking timeouts seem not to work
00036         int bytes = 0;
00037         bool first = true;
00038         do 
00039         {
00040             if (first)
00041                 first = false;
00042             else
00043                 wait_us((timeout*1000) < 100 ? (timeout*1000) : 100);
00044             int rc;
00045             if (read)
00046                 rc = mysock.recv((char*)buffer, len);
00047             else
00048                 rc = mysock.send((char*)buffer, len);
00049             if (rc < 0)
00050             {
00051                 if (rc != NSAPI_ERROR_WOULD_BLOCK)
00052                 {
00053                     bytes = -1;
00054                     break;
00055                 }
00056             } 
00057             else
00058                 bytes += rc;
00059         }
00060         while (bytes < len && timer.read_ms() < timeout);
00061         timer.stop();
00062         return bytes;
00063     }
00064 
00065     /* returns the number of bytes read, which could be 0.
00066        -1 if there was an error on the socket
00067     */
00068     int read(unsigned char* buffer, int len, int timeout)
00069     {
00070         return common(buffer, len, timeout, true);
00071     }
00072     
00073     
00074     /*int read(unsigned char* buffer, int len, int timeout)
00075     {
00076         mysock.set_timeout(timeout);
00077         int rc = mysock.recv((char*)buffer, len);
00078         mysock.set_timeout(0);
00079         return rc;
00080     }
00081 */
00082 
00083     int write(unsigned char* buffer, int len, int timeout)
00084     {
00085         return common(buffer, len, timeout, false);
00086     }
00087 
00088     int disconnect()
00089     {
00090         open = false;
00091         return mysock.close();
00092     }
00093 
00094     /*bool is_connected()
00095     {
00096         return mysock.is_connected();
00097     }*/
00098 
00099 private:
00100 
00101     bool open;
00102     TCPSocket mysock;
00103     EthernetInterface *net;
00104     Timer timer;
00105 
00106 };
00107 
00108 #endif