Version 2 of the Library. Now supporting streaming resources to websockets!! Required for real-time dashboards.

alvarolb · alvarolb · commit f2ff9bae74ba · 2015-11-13T20:27:29.000+01:00
diff --git a/library.properties b/library.properties
@@ -1,9 +1,9 @@
 name=thinger.io
-version=1.1.1
+version=2.0.0
 author=Alvaro Luis Bustamante <alvarolb@gmail.com>
 maintainer=Thinger.io <admin@thinger.io>
 sentence=Arduino library for the thinger.io IoT platform.
-paragraph=Thinger.io is an open source platform for the Internet of Things. It will allow easily connecting your things or devices for remote sensing and actuating. Working with several devices like ESP8266, Arduino Ethernet, Arduino Wifi, Arduino Yun, Adafruit CC3000, Texas Instruments CC3200.
+paragraph=Thinger.io is an open source platform for the Internet of Things. It will allow connecting your things or devices for remote sensing and actuating. Working with several devices like ESP8266, Arduino Ethernet, Arduino Wifi, Arduino Yun, Adafruit CC3000, Texas Instruments CC3200.
 category=Communication
 url=https://github.com/thinger-io/Arduino-Library
 architectures=*
diff --git a/src/ThingerWifi.h b/src/ThingerWifi.h
@@ -78,7 +78,7 @@ class ThingerWifi : public ThingerClient {
 
 public:
 
-    void add_wifi(char* ssid, char* password)
+    void add_wifi(const char* ssid, const char* password)
     {
         wifi_ssid_ = ssid;
         wifi_password_ = password;
@@ -87,8 +87,8 @@ class ThingerWifi : public ThingerClient {
 private:
 
     WiFiClient client_;
-    char* wifi_ssid_;
-    char* wifi_password_;
+    const char* wifi_ssid_;
+    const char* wifi_password_;
 };
 
 #endif
diff --git a/src/thinger/pson.h b/src/thinger/pson.h
@@ -156,6 +156,29 @@ namespace protoson {
             clear();
         }
 
+        size_t size() const{
+            size_t size = 0;
+            list_item* current = item_;
+            while(current!=NULL){
+                current = current->next_;
+                size++;
+            }
+            return size;
+        }
+
+        T* operator[](size_t index){
+            list_item* current = item_;
+            size_t current_index = 0;
+            while(current!=NULL){
+                if(current_index==index){
+                    return &current->item_;
+                }
+                current = current->next_;
+                current_index++;
+            }
+            return NULL;
+        }
+
         void clear(){
             list_item* current = item_;
             while(current!=NULL){
diff --git a/src/thinger/thinger.h b/src/thinger/thinger.h
@@ -63,7 +63,27 @@ namespace thinger{
         /**
          * Can be override to start reconnection process
          */
-        virtual void disconnected(){}
+        virtual void disconnected(){
+            // stop all streaming resources after disconnect
+            if(thinger_resource::get_streaming_counter()>0) {
+                thinger_map<thinger_resource>::entry* current = resources_.begin();
+                while(current!=NULL){
+                    current->value_.disable_streaming();
+                    current = current->next_;
+                }
+            }
+        }
+
+        /**
+         * Stream a given resource
+         */
+        void stream_resource(thinger_resource& resource, thinger_message::signal_flag type){
+            thinger_message message;
+            message.set_stream_id(resource.get_stream_id());
+            message.set_signal_flag(type);
+            resource.fill_api_io(message.get_data());
+            send_message(message);
+        }
 
     public:
 
@@ -77,7 +97,8 @@ namespace thinger{
             keep_alive_response = true;
 
             thinger_message message;
-            message.resources().add("login").add(username).add(device_id).add(credential);
+            message.set_signal_flag(thinger_message::AUTH);
+            message.resources().add(username).add(device_id).add(credential);
             if(!send_message(message)) return false;
 
             thinger_message response;
@@ -86,22 +107,44 @@ namespace thinger{
 
         bool call_endpoint(const char* endpoint_name){
             thinger_message message;
-            message.resources().add("ep").add(endpoint_name);
+            message.set_signal_flag(thinger_message::CALL_ENDPOINT);
+            message.resources().add(endpoint_name);
             return send_message(message);
         }
 
         bool call_endpoint(const char* endpoint_name, pson& data){
             thinger_message message;
+            message.set_signal_flag(thinger_message::CALL_ENDPOINT);
             message.set_data(data);
-            message.resources().add("ep").add(endpoint_name);
+            message.resources().add(endpoint_name);
             return send_message(message);
         }
 
+        bool call_endpoint(const char* endpoint_name, thinger_resource& resource){
+            thinger_message message;
+            message.set_signal_flag(thinger_message::CALL_ENDPOINT);
+            message.resources().add(endpoint_name);
+            resource.fill_api_io(message.get_data());
+            return send_message(message);
+        }
+
+        /**
+         * Stream the given resource. This resource should be previously requested by an external process.
+         * Otherwise, the resource will not be streamed as nothing will be listening for it.
+         */
+        bool stream(thinger_resource& resource){
+            if(resource.stream_enabled()){
+                stream_resource(resource, thinger_message::STREAM_EVENT);
+                return true;
+            }
+            return false;
+        }
+
         bool send_message(thinger_message& message)
         {
             thinger_encoder sink;
             sink.encode(message);
-            encoder.pb_encode_varint(message.get_message_type());
+            encoder.pb_encode_varint(MESSAGE);
             encoder.pb_encode_varint(sink.bytes_written());
             encoder.encode(message);
             return write(NULL, 0, true);
@@ -119,24 +162,35 @@ namespace thinger{
                 if(keep_alive_response){
                     last_keep_alive = current_time;
                     keep_alive_response = false;
-                    encoder.pb_encode_varint(thinger_message::KEEP_ALIVE);
+                    encoder.pb_encode_varint(KEEP_ALIVE);
                     encoder.pb_encode_varint(0);
                     write(NULL, 0, true);
                 }else{
                     disconnected();
                 }
             }
+
+            // handle streaming resources
+            if(thinger_resource::get_streaming_counter()>0){
+                thinger_map<thinger_resource>::entry* current = resources_.begin();
+                while(current!=NULL){
+                    if(current->value_.stream_required(current_time)){
+                        stream_resource(current->value_, thinger_message::STREAM_SAMPLE);
+                    }
+                    current = current->next_;
+                }
+            }
         }
 
         bool read_message(thinger_message& message){
-            uint8_t message_type = decoder.pb_decode_varint32();
-            switch (message_type){
-                case thinger_message::MESSAGE: {
+            uint8_t type = decoder.pb_decode_varint32();
+            switch (type){
+                case MESSAGE: {
                     size_t size = decoder.pb_decode_varint32();
                     decoder.decode(message, size);
                 }
                     break;
-                case thinger_message::KEEP_ALIVE: {
+                case KEEP_ALIVE: {
                     size_t size = decoder.pb_decode_varint32();
                     keep_alive_response = true;
                 }
diff --git a/src/thinger/thinger_decoder.hpp b/src/thinger/thinger_decoder.hpp
@@ -43,11 +43,13 @@ namespace thinger{
                         uint32_t size = pb_decode_varint32();
                         void *data = NULL;
                         switch (field_number) {
+                            /*
                             case thinger_message::THING_ID:
                                 data = protoson::pool.allocate(size + 1);
                                 pb_read_string((char *) data, size);
                                 message.set_thing_id((const char *) data);
                                 break;
+                            */
                             default:
                                 pb_skip(size);
                                 break;
@@ -73,7 +75,7 @@ namespace thinger{
                             case thinger_message::RESOURCE:
                                 protoson::pson_decoder::decode(message.get_resources());
                                 break;
-                            case thinger_message::PSON:
+                            case thinger_message::PSON_PAYLOAD:
                                 protoson::pson_decoder::decode(((protoson::pson&) message));
                                 break;
                             default:
@@ -116,8 +118,12 @@ namespace thinger{
 
     protected:
         virtual bool read(void* buffer, size_t size){
-            memcpy(buffer, buffer_ + read_, size);
-            return protoson::pson_decoder::read(buffer, size);
+            if(read_+size<=size_){
+                memcpy(buffer, buffer_ + read_, size);
+                return protoson::pson_decoder::read(buffer, size);
+            }else{
+                return false;
+            }
         }
 
     private:
diff --git a/src/thinger/thinger_encoder.hpp b/src/thinger/thinger_encoder.hpp
@@ -44,15 +44,12 @@ class thinger_encoder : public protoson::pson_encoder{
         if(message.get_signal_flag()!=thinger_message::NONE){
             pb_encode_varint(thinger_message::SIGNAL_FLAG, message.get_signal_flag());
         }
-        if(message.get_thing_id()!=NULL){
-            pb_encode_string(message.get_thing_id(), thinger_message::THING_ID);
-        }
         if(message.has_resource()){
             pb_encode_tag(protoson::pson_type, thinger_message::RESOURCE);
             protoson::pson_encoder::encode(message.get_resources());
         }
         if(message.has_data()){
-            pb_encode_tag(protoson::pson_type, thinger_message::PSON);
+            pb_encode_tag(protoson::pson_type, thinger_message::PSON_PAYLOAD);
             protoson::pson_encoder::encode((protoson::pson&) message);
         }
     }
diff --git a/src/thinger/thinger_message.hpp b/src/thinger/thinger_message.hpp
@@ -28,55 +28,99 @@
 
 namespace thinger{
 
+    enum message_type{
+        MESSAGE             = 1,
+        KEEP_ALIVE          = 2
+    };
+
     class thinger_message{
+
     public:
-        enum message_type{ MESSAGE = 1, KEEP_ALIVE = 2 };
-        enum signal_flag { NONE=0, REQUEST_OK = 1, REQUEST_ERROR = 2 };
-        enum fields{ STREAM_ID = 1, SIGNAL_FLAG = 2, THING_ID = 3, RESOURCE = 4, ACTION = 5, PSON = 6};
+
+        // fields for a thinger message (encoded as in protocol buffers)
+        enum fields{
+            STREAM_ID       = 1,
+            SIGNAL_FLAG     = 2,
+            UNUSED          = 3,
+            RESOURCE        = 4,
+            UNUSED2         = 5,
+            PSON_PAYLOAD    = 6
+        };
+
+        // flags for describing a thinger message
+        enum signal_flag {
+            // GENERAL USED FLAGS
+            REQUEST_OK          = 1,    // the request with the given stream id was successful
+            REQUEST_ERROR       = 2,    // the request with the given stream id failed
+
+            // SENT BY THE SERVER
+            NONE                = 0,    // default resource action: just execute the given resource
+            START_STREAM        = 3,    // enable a streaming resource (with stream_id, and resource filled, sample interval (in payload) is optional)
+            STOP_STREAM         = 4,    // stop the streaming resource (with stream_id, and resource filled)
+
+            // SENT BY DEVICE
+            AUTH                = 5,
+            STREAM_EVENT        = 6,    // means that the message data is related to a stream event
+            STREAM_SAMPLE       = 7,    // means that the message is related to a periodical streaming sample
+            CALL_ENDPOINT       = 8     // call the endpoint with the provided name (endpoint in resource, value passed in payload)
+        };
 
     public:
 
-        thinger_message(thinger_message& other) : message_type_(other.message_type_), stream_id(other.stream_id), flag(REQUEST_OK), thing_id(NULL), resource(NULL), data(NULL), data_allocated(false){
-        }
+        /**
+         * Initialize a default response  message setting the same stream id of the source message,
+         * and initializing the signal flag to ok. All remaining data or fields are empty
+         */
+        thinger_message(thinger_message& other) :
+            stream_id(other.stream_id),
+            flag(REQUEST_OK),
+            resource(NULL),
+            data(NULL),
+            data_allocated(false)
+        {}
 
-        thinger_message() : message_type_(MESSAGE), stream_id(0), flag(NONE), thing_id(NULL), resource(NULL), data(NULL), data_allocated(false)
+        /**
+         * Initialize a default empty message
+         */
+        thinger_message() :
+                stream_id(0),
+                flag(NONE),
+                resource(NULL),
+                data(NULL),
+                data_allocated(false)
         {}
 
         ~thinger_message(){
-            protoson::pool.deallocate(thing_id);
+            // deallocate resource
             destroy(resource, protoson::pool);
+            // deallocate paylaod if was allocated here
             if(data_allocated){
                 destroy(data, protoson::pool);
             }
         }
 
     private:
-        message_type message_type_;
-        uint32_t stream_id;
+        /// used for identifying a unique stream
+        uint16_t stream_id;
+        /// used for setting a stream signal
         signal_flag flag;
-        char* thing_id;
+        /// used to identify a device resource
         protoson::pson* resource;
+        /// used to fill a data payload in the message
         protoson::pson* data;
+        /// flag to determine when the payload has been reserved
         bool data_allocated;
 
     public:
 
-        message_type get_message_type(){
-            return message_type_;
-        }
-
-        uint32_t get_stream_id(){
+        uint16_t get_stream_id(){
             return stream_id;
         }
 
         signal_flag get_signal_flag(){
             return flag;
         }
 
-        const char* get_thing_id(){
-            return thing_id;
-        }
-
         bool has_data(){
             return data!=NULL;
         }
@@ -86,24 +130,14 @@ namespace thinger{
         }
 
     public:
-        void set_stream_id(uint32_t stream_id) {
+        void set_stream_id(uint16_t stream_id) {
             thinger_message::stream_id = stream_id;
         }
 
         void set_signal_flag(signal_flag const &flag) {
             thinger_message::flag = flag;
         }
 
-        void set_thing_id(const char *thing_id) {
-            size_t str_size = strlen(thing_id);
-            this->thing_id = (char*)protoson::pool.allocate(str_size+1);
-            memcpy(this->thing_id, thing_id, str_size+1);
-        }
-
-        void set_message_type(message_type type){
-            message_type_ = type;
-        }
-
     public:
 
         void operator=(const char* str){
diff --git a/src/thinger/thinger_resource.hpp b/src/thinger/thinger_resource.hpp

Original file line number	Diff line number	Diff line change
`@@ -44,15 +44,12 @@ class thinger_encoder : public protoson::pson_encoder{`
`44`	`44`	`if(message.get_signal_flag()!=thinger_message::NONE){`
`45`	`45`	`pb_encode_varint(thinger_message::SIGNAL_FLAG, message.get_signal_flag());`
`46`	`46`	`}`
`47`		`- if(message.get_thing_id()!=NULL){`
`48`		`- pb_encode_string(message.get_thing_id(), thinger_message::THING_ID);`
`49`		`- }`
`50`	`47`	`if(message.has_resource()){`
`51`	`48`	`pb_encode_tag(protoson::pson_type, thinger_message::RESOURCE);`
`52`	`49`	`protoson::pson_encoder::encode(message.get_resources());`
`53`	`50`	`}`
`54`	`51`	`if(message.has_data()){`
`55`		`- pb_encode_tag(protoson::pson_type, thinger_message::PSON);`
	`52`	`+ pb_encode_tag(protoson::pson_type, thinger_message::PSON_PAYLOAD);`
`56`	`53`	`protoson::pson_encoder::encode((protoson::pson&) message);`
`57`	`54`	`}`
`58`	`55`	`}`