stream.hpp

 
#pragma once
 
#include "../asrtl/asrt_assert.h"
#include "../asrtl/log.h"
#include "../asrtl/util.h"
#include "../asrtlpp/callback.hpp"
#include "../asrtlpp/task.hpp"
#include "../asrtlpp/util.hpp"
#include "../asrtr/stream.h"
#include "./task_unit.hpp"
 
namespace asrt
{
 
template < typename T >
struct strm_field_traits;
 
template <>
struct strm_field_traits< uint8_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_U8;
        static constexpr auto size = 1;
        static void           encode( uint8_t*& p, uint8_t v ) { *p++ = v; }
};
 
template <>
struct strm_field_traits< uint16_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_U16;
        static constexpr auto size = 2;
        static void           encode( uint8_t*& p, uint16_t v ) { asrt_add_u16( &p, v ); }
};
 
template <>
struct strm_field_traits< uint32_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_U32;
        static constexpr auto size = 4;
        static void           encode( uint8_t*& p, uint32_t v ) { asrt_add_u32( &p, v ); }
};
 
template <>
struct strm_field_traits< int8_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_I8;
        static constexpr auto size = 1;
        static void encode( uint8_t*& p, int8_t v ) { *p++ = static_cast< uint8_t >( v ); }
};
 
template <>
struct strm_field_traits< int16_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_I16;
        static constexpr auto size = 2;
        static void           encode( uint8_t*& p, int16_t v )
        {
                asrt_add_u16( &p, static_cast< uint16_t >( v ) );
        }
};
 
template <>
struct strm_field_traits< int32_t >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_I32;
        static constexpr auto size = 4;
        static void           encode( uint8_t*& p, int32_t v ) { asrt_add_i32( &p, v ); }
};
 
template <>
struct strm_field_traits< float >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_FLOAT;
        static constexpr auto size = 4;
        static void           encode( uint8_t*& p, float v )
        {
                uint32_t u;
                std::memcpy( &u, &v, 4 );
                asrt_add_u32( &p, u );
        }
};
 
template <>
struct strm_field_traits< bool >
{
        static constexpr auto tag  = ASRT_STRM_FIELD_BOOL;
        static constexpr auto size = 1;
        static void           encode( uint8_t*& p, bool v ) { *p++ = v ? 1 : 0; }
};
 
ASRT_NODISCARD inline status init( ref< asrt_stream_client > client, asrt_node& prev )
{
        return asrt_stream_client_init( client, &prev );
}
 
ASRT_NODISCARD inline status define(
    ref< asrt_stream_client >          client,
    uint8_t                            schema_id,
    enum asrt_strm_field_type_e const* fields,
    uint8_t                            field_count,
    callback< asrt_stream_done_cb >    done_cb )
{
        return asrt_stream_client_define(
            client, schema_id, fields, field_count, done_cb.fn, done_cb.ptr );
}
ASRT_NODISCARD inline status emit(
    ref< asrt_stream_client >       client,
    uint8_t                         schema_id,
    uint8_t const*                  data,
    uint16_t                        data_size,
    callback< asrt_stream_done_cb > done_cb )
{
        return asrt_stream_client_emit(
            client, schema_id, data, data_size, done_cb.fn, done_cb.ptr );
}
 
ASRT_NODISCARD inline status reset( ref< asrt_stream_client > client )
{
        return asrt_stream_client_reset( client );
}
 
inline void deinit( ref< asrt_stream_client > client )
{
        asrt_stream_client_deinit( client );
}
 
template < typename... Ts >
struct stream_schema
{
        static constexpr uint16_t emit_size = ( strm_field_traits< Ts >::size + ... );
 
        stream_schema(
            ref< asrt_stream_client >       client,
            uint8_t                         schema_id,
            callback< asrt_stream_done_cb > done_cb )
          : _client( client )
          , _schema_id( schema_id )
        {
                auto s = define( _client, _schema_id, fields, sizeof...( Ts ), done_cb );
                if ( s != ASRT_SUCCESS ) {
                        ASRT_ERR_LOG( "asrt_stream_schema", "define failed" );
                        ASRT_ASSERT( false );
                }
        }
 
        stream_schema( ref< asrt_stream_client > c, uint8_t id )
          : _client( c )
          , _schema_id( id )
        {
        }
 
 
        stream_schema( stream_schema const& )            = delete;
        stream_schema& operator=( stream_schema const& ) = delete;
 
        stream_schema( stream_schema&& o ) noexcept
          : _client( o._client )
          , _schema_id( o._schema_id )
        {
                o._client = nullptr;
        }
 
        stream_schema& operator=( stream_schema&& o ) noexcept
        {
                _client    = o._client;
                _schema_id = o._schema_id;
                o._client  = nullptr;
                return *this;
        }
 
        ASRT_NODISCARD status emit( Ts... args, callback< asrt_stream_done_cb > done_cb )
        {
                uint8_t* p = _emit_buf;
                ( strm_field_traits< Ts >::encode( p, args ), ... );
                return asrt::emit( _client, _schema_id, _emit_buf, emit_size, done_cb );
        }
 
        ASRT_NODISCARD status
        emit_raw( uint8_t const* buf, callback< asrt_stream_done_cb > done_cb )
        {
                return asrt::emit( _client, _schema_id, buf, emit_size, done_cb );
        }
 
        ~stream_schema() = default;
 
        static constexpr enum asrt_strm_field_type_e fields[] = { strm_field_traits< Ts >::tag... };
 
private:
        asrt_stream_client* _client;
        uint8_t             _schema_id;
        uint8_t             _emit_buf[emit_size > 0 ? emit_size : 1];
};
 
template < typename... Ts >
struct _stream_define_ctx
{
        using completion_signatures = ecor::completion_signatures<
            ecor::set_value_t( stream_schema< Ts... > ),
            ecor::set_error_t( status ) >;
 
        asrt_stream_client* client;
        uint8_t             schema_id;
 
        template < typename OP >
        void start( OP& op )
        {
                auto cb = +[]( void* ptr, enum asrt_status s ) {
                        auto& o = *static_cast< OP* >( ptr );
                        if ( s == ASRT_SUCCESS )
                                o.receiver.set_value(
                                    stream_schema< Ts... >{ o.ctx.client, o.ctx.schema_id } );
                        else
                                o.receiver.set_error( s );
                };
                auto s = define(
                    *client,
                    schema_id,
                    stream_schema< Ts... >::fields,
                    sizeof...( Ts ),
                    { cb, &op } );
                if ( s != ASRT_SUCCESS )
                        op.receiver.set_error( s );
        }
};
 
template < typename... Ts >
using stream_define_sender = ecor::sender_from< _stream_define_ctx< Ts... > >;
 
template < typename... Ts >
ecor::sender auto define( asrt_stream_client& client, uint8_t schema_id )
{
        return stream_define_sender< Ts... >{ { &client, schema_id } };
}
 
 
template < typename... Ts >
struct _stream_emit_ctx
{
        using completion_signatures =
            ecor::completion_signatures< ecor::set_value_t(), ecor::set_error_t( status ) >;
 
        stream_schema< Ts... >* schema;
        uint8_t                 buf[stream_schema< Ts... >::emit_size];
 
        template < typename OP >
        void start( OP& op )
        {
                auto cb = +[]( void* ptr, enum asrt_status s ) {
                        auto& o = *static_cast< OP* >( ptr );
                        if ( s == ASRT_SUCCESS )
                                o.receiver.set_value();
                        else
                                o.receiver.set_error( s );
                };
                auto st = schema->emit_raw( buf, { cb, &op } );
                if ( st != ASRT_SUCCESS )
                        op.receiver.set_error( st );
        }
};
 
template < typename... Ts >
using stream_emit_sender = ecor::sender_from< _stream_emit_ctx< Ts... > >;
 
template < typename... Ts >
ecor::sender auto emit( stream_schema< Ts... >& schema, Ts... args )
{
        _stream_emit_ctx< Ts... > ctx{ &schema, {} };
        uint8_t*                  p = ctx.buf;
        ( strm_field_traits< Ts >::encode( p, args ), ... );
        return stream_emit_sender< Ts... >{ std::move( ctx ) };
}
 
 
}  // namespace asrt