ADD: Message container and Join Message

CMakeLists.txt

@@ -41,6 +41,19 @@ add_library(whisper-com STATIC
   include/WHISPER/whisper.hpp
   src/WHISPER/whisper.cpp
   
+  include/WHISPER/threadSafeQueue.hpp
+  src/WHISPER/threadSafeQueue.cpp
+
+   
+  include/WHISPER/Messages/Message.hpp
+  src/WHISPER/Messages/Message.cpp
+
+  include/WHISPER/Messages/Join.hpp
+  src/WHISPER/Messages/Join.cpp
+ 
+  
+  ../include/WHISPER/Messages/Protos/message.pb.cc           
+
 )
 
 

include/WHISPER/Messages/Join.hpp

@@ -0,0 +1,24 @@
+#pragma once
+
+
+#include <WHISPER/Messages/Message.hpp>
+#include <WHISPER/Messages/Protos/join.pb.h>
+#include <string>
+#include <loguru.hpp>
+
+namespace WHISPER {
+
+    class Join : public Message
+    {             
+        private:
+            messages::join::Join joinMessage;
+        public:
+            std::uint32_t port;
+            std::string sourceAddr;
+            
+            Join(std::string receivedMessage);
+
+            Join(std::uint32_t deviceID,std::uint32_t topic, MsgType Type, SourceType src,std::uint32_t port, std::string addr);
+
+    };
+}

include/WHISPER/Messages/Message.hpp

@@ -0,0 +1,88 @@
+#pragma once
+
+
+
+
+#include "google/protobuf/any.pb.h"
+#include "google/protobuf/message.h"
+#include <WHISPER/Messages/Protos/message.pb.h>
+#include <memory>
+#include <string>
+#include <loguru.hpp>
+
+
+namespace WHISPER {
+
+   enum  MsgType : int32_t
+    {   
+    /// message to shutdown all participants
+    SHUTDOWN = 0,
+    /// on participant is joined
+    JOIN=1,
+    /// participant is leaving
+    LEAVE,
+    /// owntrack informaton
+    OWN_TRACK,
+    /// raw track message
+    RAW_TRACK,
+    /// simple data
+    SIMPLE
+    }; // enum class EventType
+
+     enum SourceType : int32_t 
+    {   
+    SIMCOMTROLER,    
+    /// 
+    SHIP,
+    /// 
+    SENSOR,
+    /// 
+    EFEKTOR,
+    /// 
+    GATEWAY,
+   /// 
+    
+    
+    }; // enum class EventType
+
+    
+
+
+    class Message{  
+        private:
+            std::string payloadString_;     
+            std::shared_ptr<google::protobuf::Any> payload_; 
+
+        public:
+            Message()=default;
+
+            Message(std::int32_t deviceId, std::uint32_t topic, MsgType Type,SourceType src);
+            Message(std::string msg);
+
+            std::string getPayloadString();
+
+            std::uint32_t topic_;
+            std::int32_t msgType_;
+            std::int32_t sourceType_;
+            std::int32_t deviceId_;
+
+            void addPayLoad(std::shared_ptr<google::protobuf::Any> any);
+            void addPayLoad(std::string any);
+
+            std::string serialize();
+
+            
+        protected:    
+            
+            messages::header::Message msg;
+           
+    };
+
+
+ 
+
+
+
+
+
+}

include/WHISPER/Messages/Protos/join.proto

@@ -3,14 +3,13 @@
 syntax = "proto3";
 package messages.join;
 
-import "google/protobuf/timestamp.proto";
+// import "google/protobuf/timestamp.proto";
 // [END declaration]
 
 
 // [START messages]
 message Join {
-  int32 id = 1;
-  int32 port = 2;  
-  string address = 3;
+  uint32 port = 1;  
+  string srcAddress = 2;
 }
 

include/WHISPER/Messages/Protos/message.proto

@@ -0,0 +1,19 @@
+syntax = "proto3";
+
+
+import "google/protobuf/any.proto";
+
+package messages.header;
+
+// [START messages]
+message Message {
+  int32 topic = 1;
+  int32 msgType = 2;
+  int32 sourceType = 3; 
+  int32 sourceID = 4;  
+
+  repeated google.protobuf.Any payload = 5; 
+
+}
+
+

include/WHISPER/threadSafeQueue.hpp

@@ -0,0 +1,69 @@
+#pragma once
+/* This Source Code Form is subject to the terms of the Mozilla Public
+* License, v. 2.0. If a copy of the MPL was not distributed with this
+* file, You can obtain one at https://mozilla.org/MPL/2.0/.
+*/
+
+/**
+* @file
+* @copyright MPLv2
+*/
+#include <mutex>
+#include <queue>
+#include <memory>
+#include <condition_variable>
+#include <string>
+
+namespace WHISPER{
+  /**
+    * @class BasicMessageQueue
+    *
+    * This class encapsulates shared storage which is a queue and protect it
+    * with a mutex. In Addition there is a condition variable to notify other
+    * threads if there is new data in the queue.
+    */
+  template<class T>
+  class threadSafeQueue
+  {
+  private:
+    std::queue<std::unique_ptr<T>> q;
+    std::mutex mx;
+    std::condition_variable condVar;
+  public:
+    /**
+      * @brief default constructor for class BasicMessageQueue
+      */
+    threadSafeQueue();
+
+    /**
+      * @brief appends given message to queue
+      *
+      * The storage is protected by a mutex. Makes a notify_one call to inform 
+      * in waiting thread that there is a new message in the queue.
+      *
+      * @param msg - incoming message
+      */
+    void addElement( std::unique_ptr<T> msg );
+
+    /**
+      * @brief gets the fron message from the queue
+      *
+      * This method gets the front message in the queue and deletes it from the 
+      * queue. The storage is protected by a mutex. If the queue is empty the
+      * function throws an exception.
+      *
+      * @result msg - returns the front message from the queue
+      * @throws length_error - if queue is empty this method throws length_error exception
+      */
+    std::unique_ptr<T> getElement();
+
+    /**
+      * @brief method size
+      *
+      * @result size - current size of the message queue
+      */
+    unsigned int size();
+
+  }; //BasicMessageQueue
+
+}; // namespace BC

include/WHISPER/whisper.hpp

@@ -10,8 +10,21 @@
 * @copyright 2022 MPLv2
 */
 
+#include <cstddef>
+#include <cstdint>
+#include <memory>
 #define ZMQ_BUILD_DRAFT_API 1
 #include <zmq.hpp>
+#include <string>
+#include <vector>
+#include <atomic>
+#include <thread>
+
+#include <loguru.hpp>
+
+
+#include <WHISPER/threadSafeQueue.hpp>
+#include <WHISPER/Messages/Message.hpp>
 
 /**
   * @brief namespace for all whisper-com related components
@@ -21,6 +34,56 @@ namespace WHISPER
   // Add datatypes here
   class whispercomm{
 
+    private: 
+      /// device ID
+      std::uint32_t ownID_;
+      /// device Type
+      SourceType ownDeviceType_;
+      /// all topics this service subscribed
+      std::vector<std::string> subscribedTopics;
+     
+      /// show if the service is connected or not
+      std::atomic<bool> connected;
+     
+      /// attribute identifying this service as a gateway and all packets should be forwarded
+      bool gateway = false;
+
+      /// variable for holding the receive thread identifier
+      std::thread receiveThread;
+
+      /// variable indicating if the receiveThread should be stopped
+      std::atomic<bool> stopReceiveThread;
+
+      std::shared_ptr<threadSafeQueue<std::string>> receiveQueue = nullptr;
+      
+      
+      
+      void receive();
+
+
+    public:     
+      whispercomm(std::uint32_t id, SourceType owndevicetype):ownID_(id),ownDeviceType_(owndevicetype)
+      {
+        receiveQueue = std::make_shared<threadSafeQueue<std::string>>();
+      };
+      void connect(std::shared_ptr<threadSafeQueue<std::string>> receiver);
+      void publish(std::string msg);
+      void disconnect();
+      void subscribe(std::string topic);
+      void unsubscribe(std::string topic);
+
+    protected:
+
+    void addMsgToReceiverQueue(std::string);
+
+
+    virtual void derivedConnect() = 0;
+    virtual void derivedDisconnect() = 0;
+    virtual void derivedPublish(std::string msg) = 0;
+    virtual void derivedReceive() = 0;
+
+
+
 
 
 

src/WHISPER/Messages/Join.cpp

@@ -0,0 +1,53 @@
+#include "WHISPER/Messages/Message.hpp"
+#include "WHISPER/Messages/Protos/join.pb.h"
+#include <WHISPER/Messages/Join.hpp>
+#include <memory>
+
+
+
+
+namespace WHISPER {
+
+        Join::Join(std::string receivedMessage){
+             msg = messages::header::Message();
+        try {
+            msg.ParseFromString(receivedMessage);
+            topic_ = msg.topic();
+            sourceType_ = msg.sourcetype();
+            msgType_ = msg.msgtype();
+            joinMessage = messages::join::Join();
+            if ( msg.payload_size()) {
+                if (msg.payload().begin()->Is<messages::join::Join>()) {
+                    msg.payload().begin()->UnpackTo(&joinMessage);
+                }     
+            }
+            port = joinMessage.port();
+            sourceAddr = joinMessage.srcaddress();
+           
+            
+
+        } catch (const std::exception& e) {
+            LOG_S(ERROR)<<e.what();
+        }
+
+        }
+       
+
+        WHISPER::Join::Join(std::uint32_t deviceID,std::uint32_t topic, MsgType Type, SourceType src,std::uint32_t port, std::string addr):
+        Message(deviceID,topic,Type,src),port(port),sourceAddr(addr)
+        {
+            joinMessage = messages::join::Join();
+            joinMessage.set_port(port);
+            joinMessage.set_srcaddress(sourceAddr);
+
+            LOG_S(INFO)<< "join message befor packing:" << joinMessage.ByteSizeLong();
+            auto test = std::make_shared<google::protobuf::Any>();
+            test->PackFrom(joinMessage);
+            addPayLoad(test);
+
+        }
+
+
+        
+
+}

src/WHISPER/Messages/Message.cpp

@@ -0,0 +1,72 @@
+
+
+#include "WHISPER/Messages/Protos/join.pb.h"
+#include <WHISPER/Messages/Message.hpp>
+#include <memory>
+
+
+
+
+namespace WHISPER {
+    Message::Message(std::string stringMessage)
+    {
+        msg = messages::header::Message();
+        try {
+            msg.ParseFromString(stringMessage);
+            topic_ = msg.topic();
+            sourceType_ = msg.sourcetype();
+            msgType_ = msg.msgtype();
+
+
+        } catch (const std::exception& e) {
+            LOG_S(ERROR)<<e.what();
+        }
+        
+
+    }
+
+
+     Message::Message(std::int32_t deviceId,std::uint32_t topic, MsgType Type,SourceType src):topic_(topic),sourceType_(src),msgType_(Type){
+        msg =  messages::header::Message();
+        
+        if(msg.IsInitialized())
+        {   
+            msg.set_sourceid(deviceId);
+            msg.set_topic(topic);
+            msg.set_sourcetype(sourceType_);
+            msg.set_msgtype(msgType_);
+        }
+     }
+
+   
+
+    std::string Message::getPayloadString(){
+        return payloadString_;
+    }
+
+    
+     void Message::addPayLoad(std::shared_ptr<google::protobuf::Any> payload){
+        payload_ = payload;
+        LOG_S(INFO)<< "pack any size in message class "<<payload_->ByteSizeLong();
+
+        msg.add_payload()->CopyFrom(*payload_);
+        
+        LOG_S(INFO)<< "pack any size in message class "<<payload_->ByteSizeLong();
+
+    }
+
+    
+
+    std::string Message::serialize(){
+            
+            std::string serializedMessage;
+            LOG_S(INFO)<<msg.ByteSizeLong();
+            if (msg.IsInitialized()) {
+                serializedMessage = msg.SerializeAsString();
+            }
+
+            return serializedMessage;
+        }
+
+    
+}

src/WHISPER/threadSafeQueue.cpp

@@ -0,0 +1,60 @@
+/*
+* This Source Code Form is subject to the terms of the Mozilla Public
+* License, v. 2.0. If a copy of the MPL was not distributed with this
+* file, You can obtain one at https://mozilla.org/MPL/2.0/.
+*/
+
+/**
+* @file BasicMessageQueue.cpp
+* @brief class which encapsulates queue
+* @author Christina Sander <christina.sander@hsu-hh.de>
+* @date   25.06.2020
+* @copyright MPLv2
+*/
+#include <WHISPER/threadSafeQueue.hpp>
+#include <iostream>
+#include <string>
+
+template<class T>
+WHISPER::threadSafeQueue<T> threadSafeQueue(){
+        
+}
+
+
+
+// appends the given message to the message queue
+template<class T>
+void WHISPER::threadSafeQueue< T>::addElement(std::unique_ptr<T> elem) 
+{
+  std::unique_lock<std::mutex> lk(mx);
+  q.push( std::move(elem) );
+  lk.unlock();
+  condVar.notify_one();
+}
+
+// gets a message from the queue
+template<class T>
+std::unique_ptr<T> WHISPER::threadSafeQueue<T>::getElement()
+{
+  std::unique_lock<std::mutex> lk(mx);
+  if( 0 == q.size() )
+  {
+    lk.unlock();
+    throw std::length_error("Empty Queue\n");
+  }
+  std::unique_ptr<std::string> elem = std::move( q.front() );
+  q.pop();
+  lk.unlock();
+  return std::move(elem);
+}
+
+
+//returns the size of the message queue
+template<class T>
+unsigned int WHISPER::threadSafeQueue<T>::size()
+{
+  std::unique_lock<std::mutex> lk(mx);
+  unsigned int size = q.size();
+  lk.unlock();
+  return size;
+}

src/WHISPER/whisper.cpp

@@ -15,5 +15,59 @@
   */
 namespace WHISPER
 {
+
+
+
+  void whispercomm::connect(std::shared_ptr<threadSafeQueue<std::string>> receiver)
+  {
+    this->receiveQueue = receiver;
+    this->derivedConnect();
+    receiveThread = std::thread(&WHISPER::whispercomm::receive,this);
+
+  }
+
+  void whispercomm::disconnect()
+  {
+
+  }
+
+  void whispercomm::publish(std::string msg){
+    this->derivedPublish(msg);
+  }
+
+  void whispercomm::receive(){
+    this->derivedReceive();
+    connected = true;
+
+    while(!stopReceiveThread)
+  {
+    derivedReceive();
+  }
+   
+  }
+
+  
+  void whispercomm::subscribe(std::string topic)
+  {
+    this->subscribedTopics.push_back(topic);
+  }
+
+  void whispercomm::unsubscribe(std::string topic)
+  { 
+    for (std::vector<std::string>::iterator it = subscribedTopics.begin(); it != subscribedTopics.end();it++)
+    {
+        if (*it  == topic)
+        {
+            it = subscribedTopics.erase(it);
+        }
+    }
+
+  }
+
+  void whispercomm::addMsgToReceiverQueue(std::string)
+  {
+
+  }
+
   // Add datatypes here
 } // namespace WHISPER

src/main.cpp

@@ -1,12 +1,36 @@
+#include "WHISPER/Messages/Message.hpp"
 #include <iostream>
+#include <loguru.hpp>
 
+#include <WHISPER/Messages/Join.hpp>
+#include <WHISPER/Messages/Protos/message.pb.h>
 
-#include "../include/WHISPER/Messages/Protos/join.pb.h"
 
 int main()
 {
      GOOGLE_PROTOBUF_VERIFY_VERSION;
 
 
+
+     WHISPER::Join join(1,1,WHISPER::MsgType::JOIN,WHISPER::SourceType::SHIP,8000,"127.0.0.1");
+     std::string msg = join.serialize();
+     LOG_S(INFO)<<" serialized Message is "<<msg.size();
+     LOG_S(INFO)<<msg;
+
+     messages::header::Message proto; 
+     proto.ParseFromString(msg);
+
+     WHISPER::Message receivedMessage(msg);
+
+     LOG_S(INFO)<<receivedMessage.msgType_;
+
+     switch (receivedMessage.msgType_) {
+     case WHISPER::MsgType::JOIN:
+          WHISPER::Join receivedJoin(msg);
+          LOG_S(INFO)<< "join message data afer reception "<< receivedJoin.port; 
+     break;
+     }     
+
+
 return 0;
 }

src/protos/join.proto

@@ -1,16 +0,0 @@
-
-// [START declaration]
-syntax = "proto3";
-package messages.join;
-
-import "google/protobuf/timestamp.proto";
-// [END declaration]
-
-
-// [START messages]
-message Join {
-  int32 id = 1;
-  int32 port = 2;  
-  string address = 3;
-}
-