WebsocketServer.hh

Go to the documentation of this file.

/*
 * Copyright (C) 2019 Open Source Robotics Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
*
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
 
// This file was ported from:
// https://github.com/gazebosim/gz-launch/blob/main/plugins/websocket_server
// and converted to a gz-sim system.
 
#ifndef GZ_SIM_SYSTEMS_WEBSOCKETSERVER_HH_
#define GZ_SIM_SYSTEMS_WEBSOCKETSERVER_HH_
 
#include <chrono>
#include <cstddef>
#include <list>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <thread>
#include <vector>
 
#include <gz/transport/MessageInfo.hh>
#include <gz/transport/Node.hh>
#include <gz/common/Util.hh>
#include <libwebsockets.h>
 
#include "gz/sim/System.hh"
 
namespace gz
{
namespace sim
{
// Inline bracket to help doxygen filtering.
inline namespace GZ_SIM_VERSION_NAMESPACE {
namespace systems
{
 
    class WebsocketServer
        : public System,
          public ISystemConfigure
    {
      public: WebsocketServer() = default;
 
      public: virtual ~WebsocketServer();
 
      public: void Configure(const Entity &_entity,
                             const std::shared_ptr<const sdf::Element> &_sdf,
                             EntityComponentManager &_ecm,
                             EventManager &_eventMgr) final;
 
      public: void Run();
 
      private: void OnWebsocketSubscribedMessage(const char *_data,
                   const size_t _size,
                   const gz::transport::MessageInfo &_info);
 
      private: void OnWebsocketSubscribedImageMessage(
          const gz::msgs::Image &_msg,
          const gz::transport::MessageInfo &_info);
 
      public: void OnConnect(int _socketId);
 
      public: void OnDisconnect(int _socketId);
 
      public: void OnMessage(int _socketId, const std::string _msg);
 
      public: bool UpdateMsgTypeSubscriptionCount(const std::string &_topic,
          int _socketId, bool _subscribe);
 
      private: void OnAsset(int _socketId,
                   const std::vector<std::string> &_frameParts);
 
      private: void OnRequest(int _socketId,
                   const std::vector<std::string> &_frameParts);
 
      private: gz::transport::Node node;
 
      private: bool run = true;
      private: std::thread *thread = nullptr;
      private: struct lws_context *context = nullptr;
 
      private: std::vector<struct lws_protocols> protocols;
      private: class Connection
      {
        public: std::chrono::system_clock::time_point creationTime;
        public: std::list<std::unique_ptr<char []>> buffer;
        public: std::list<int> len;
        public: std::mutex mutex;
 
        public: bool authorized{false};
 
        public: std::map<std::string, std::chrono::nanoseconds>
            topicPublishPeriods;
 
        public: std::map<std::string,
            std::chrono::time_point<std::chrono::steady_clock>> topicTimestamps;
 
        public: std::map<std::string, int> msgTypeSubscriptionCount;
      };
 
      private: void QueueMessage(Connection *_connection,
                   const char *_data, const size_t _size);
 
      public: std::mutex mutex;
 
      public: std::mutex subscriptionMutex;
 
      public: std::map<int, std::unique_ptr<Connection>> connections;
 
      public: std::map<std::string, std::set<int>> topicConnections;
 
      public: std::map<std::string, int> msgTypeSubscriptionLimit;
 
      public: std::mutex runMutex;
 
      public: std::condition_variable runConditionVariable;
 
      public: int messageCount{0};
 
      public: int maxConnections{-1};
 
      private: std::map<std::string,
               std::chrono::time_point<std::chrono::steady_clock>>
                 topicTimestamps;
 
      public: int queueSizePerConnection{-1};
 
      private: enum Operation
               {
                 SUBSCRIBE = 0,
 
                 PUBLISH = 1,
 
                 TOPICS = 2,
 
                 PROTOS = 3,
 
                 ASSET = 4,
 
                 REQUEST = 5,
               };
 
      private: std::vector<std::string> operations{
                 "sub", "pub", "topics", "protos", "asset", "req"};
 
      private: std::map<std::string, std::string> publishHeaders;
 
      private: std::chrono::nanoseconds publishPeriod;
 
      private: std::string authorizationKey;
 
      private: std::string adminAuthorizationKey;
    };
}
}
}
}
 
#endif