stompframestream.h

Go to the documentation of this file.

/*---------------------------------------------------------------------\
|                          ____ _   __ __ ___                          |
|                         |__  / \ / / . \ . \                         |
|                           / / \ V /|  _/  _/                         |
|                          / /__ | | | | | |                           |
|                         /_____||_| |_| |_|                           |
|                                                                      |
-----------------------------------------------------------------------/
*
* This file contains private API, this might break at any time between releases.
* You have been warned!
*
*/
 
#ifndef ZYPP_CORE_ZYPPNG_RPC_STOMPFRAMESTREAM_H_INCLUDED
#define ZYPP_CORE_ZYPPNG_RPC_STOMPFRAMESTREAM_H_INCLUDED
 
#include <zypp-core/TriBool.h>
#include <zypp-core/zyppng/base/Base>
#include <zypp-core/zyppng/core/String>
#include <zypp-core/zyppng/base/Signals>
#include <zypp-core/zyppng/base/Timer>
#include <zypp-core/zyppng/io/IODevice>
#include <zypp-core/zyppng/pipelines/expected.h>
 
#include <zypp-core/rpc/PluginFrame.h>
 
#include <deque>
#include <optional>
 
namespace zyppng {
 
  ZYPP_FWD_DECL_TYPE_WITH_REFS (StompFrameStream);
 
  class ZYPP_API InvalidMessageReceivedException : public zypp::Exception
  {
  public:
    InvalidMessageReceivedException( const std::string &msg = {});
  };
 
  namespace rpc {
 
    template <typename T>
    expected<zypp::PluginFrame> toStompMessage( const T& msg ) {
      return msg.toStompMessage();
    }
 
    template <typename T>
    expected<T> fromStompMessage( const zypp::PluginFrame &message ) {
      return T::fromStompMessage( message );
    }
 
    // Reads data from the stomp message and converts it to the target type
    // used to read header values and values serialized into a terminated data field
    template <typename T>
    void parseDataIntoField( const std::string &headerVal, T &target  )
    {
      if constexpr ( std::is_same_v<bool, T> ) {
        const auto &triBool = zypp::str::strToTriBool ( headerVal );
        if ( indeterminate(triBool) ) {
          ZYPP_THROW ( zypp::PluginFrameException( "Invalid value for boolean field" ) );
        }
        target = bool(triBool);
      } else if constexpr ( std::is_same_v<std::string, T> ) {
        target = headerVal;
      } else {
        // numbers
        auto val = zyppng::str::safe_strtonum<T> ( headerVal );
        if ( !val )
          ZYPP_THROW ( zypp::PluginFrameException( "Invalid value for numerical field" ) );
        target = *val;
      }
    }
 
    // queries a header value and calls parseDataIntoField on it
    template <typename T>
    void parseHeaderIntoField( const zypp::PluginFrame &msg, const std::string &name, T &target  )
    {
      return parseDataIntoField ( msg.getHeader(name), target );
    }
 
    template <typename T>
    inline zypp::PluginFrame prepareFrame() {
      return zypp::PluginFrame ( std::string( T::typeName.data(), T::typeName.length() ) );
    }
 
  }

  class ZYPP_API StompFrameStream : public zyppng::Base
  {
    public:
 
      using Ptr = StompFrameStreamRef;

      static Ptr create( IODevice::Ptr iostr ) {
        return Ptr( new StompFrameStream( std::move(iostr) ) );
      }

      std::optional<zypp::PluginFrame> nextMessage ( const std::string &msgName = "" );

      std::optional<zypp::PluginFrame> nextMessageWait ( const std::string &msgName = "" );

      bool sendFrame ( const zypp::PluginFrame &message );
 
      template <typename T>
      bool sendMessage ( const T &message )
      {
        if constexpr ( std::is_same_v<T, zypp::PluginFrame> ) {
          return sendFrame( message );
        } else {
          const auto &msg = rpc::toStompMessage(message);
          if ( !msg ) {
            ERR << "Failed to serialize message" << std::endl;
            return false;
          }
          return sendFrame( *msg );
        }
      }
 
      template<class T>
      static expected< T > parseMessage ( const zypp::PluginFrame &m ) {
        return rpc::fromStompMessage<T>(m);
      }

      void readAllMessages ();

      SignalProxy<void()> sigMessageReceived ();

      SignalProxy<void()> sigInvalidMessageReceived ();
 
    private:
      StompFrameStream( IODevice::Ptr iostr );
      bool readNextMessage ();
      void timeout( const zyppng::Timer &);
 
      enum ParserState {
        ReceiveCommand,
        ReceiveHeaders,
        ReceiveBody,
        ParseError
      } _parserState = ReceiveCommand;
 
      std::optional<zypp::PluginFrame> _pendingMessage;
      std::optional<int64_t> _pendingBodyLen;
 
      IODevice::Ptr _ioDev;
      Timer::Ptr _nextMessageTimer = Timer::create();
      std::deque<zypp::PluginFrame> _messages;
      Signal<void()> _sigNextMessage;
      Signal<void()> _sigInvalidMessageReceived;
 
  };
}
 
#endif // ZYPP_CORE_ZYPPNG_RPC_STOMPFRAMESTREAM_H_INCLUDED