stream_transport.h

Go to the documentation of this file.

#pragma once
 
#include "fl/task/promise.h"
#include "fl/stl/function.h"
#include "fl/stl/optional.h"
#include "fl/stl/shared_ptr.h"
#include "fl/stl/span.h"
#include "fl/stl/string.h"
#include "fl/stl/flat_map.h"
#include "fl/stl/vector.h"
#include "fl/stl/asio/http/connection.h"
#include "fl/net/http/chunked_encoding.h"
#include "fl/stl/noexcept.h"
 
namespace fl {
namespace net {
namespace http {
 
class HttpStreamTransport;  // IWYU pragma: keep

class StreamHandle {
public:
    StreamHandle() FL_NOEXCEPT = default;

    StreamHandle& onData(fl::function<void(const fl::json&)> cb);

    StreamHandle& then(fl::function<void(const fl::json&)> cb);

    StreamHandle& catch_(fl::function<void(const fl::task::Error&)> cb);

    fl::task::Promise<fl::json>& promise();

    bool valid() const;
 
private:
    friend class HttpStreamTransport;
    StreamHandle(fl::task::Promise<fl::json> p,
                 fl::shared_ptr<fl::function<void(const fl::json&)>> updateCb);
 
    fl::task::Promise<fl::json> mPromise;
    fl::shared_ptr<fl::function<void(const fl::json&)>> mUpdateCallback;
};

class HttpStreamTransport {
public:
    using StateCallback = fl::function<void()>;

    HttpStreamTransport(const fl::string& host, u16 port, u32 heartbeatIntervalMs = 30000);

    virtual ~HttpStreamTransport() FL_NOEXCEPT;
 
    // Connection Management

    virtual bool connect() = 0;

    virtual void disconnect() = 0;

    virtual bool isConnected() const = 0;
 
    // RequestSource Implementation (for Remote)

    fl::optional<fl::json> readRequest();
 
    // ResponseSink Implementation (for Remote)

    void writeResponse(const fl::json& response);
 
    // Promise-based RPC API

    fl::task::Promise<fl::json> rpc(const fl::string& method, const fl::json& params);

    fl::task::Promise<fl::json> rpc(const fl::json& fullRequest);

    StreamHandle rpcStream(const fl::string& method, const fl::json& params);

    StreamHandle rpcStream(const fl::json& fullRequest);
 
    // Update Loop

    void update(u32 currentTimeMs);
 
    // Callbacks

    void setOnConnect(StateCallback callback);

    void setOnDisconnect(StateCallback callback);
 
    // Configuration

    void setHeartbeatInterval(u32 intervalMs);

    u32 getHeartbeatInterval() const;

    void setTimeout(u32 timeoutMs);

    u32 getTimeout() const;
 
protected:
    // Abstract methods for subclasses

    virtual int sendData(fl::span<const u8> data) = 0;

    virtual int recvData(fl::span<u8> buffer) = 0;

    virtual u32 getCurrentTimeMs() const;

    virtual void triggerReconnect();
 
    // Connection management
    HttpConnection mConnection;
 
private:
    // Pending call state (for rpc())
    struct PendingCall {
        fl::task::Promise<fl::json> promise;
        bool ackReceived = false;
    };
 
    // Pending stream state (for rpcStream())
    struct PendingStream {
        fl::task::Promise<fl::json> promise;
        fl::shared_ptr<fl::function<void(const fl::json&)>> updateCallback;
        bool ackReceived = false;
    };
 
    // Chunked encoding
    ChunkedReader mReader;
    ChunkedWriter mWriter;
 
    // Heartbeat tracking
    u32 mLastHeartbeatSent;
    u32 mLastHeartbeatReceived;
    u32 mHeartbeatInterval;
    u32 mTimeoutMs;
 
    // Connection state
    bool mWasConnected;
 
    // Callbacks
    StateCallback mOnConnect;
    StateCallback mOnDisconnect;
 
    // Promise-based call tracking
    fl::flat_map<fl::string, PendingCall, fl::StringFastLess> mPendingCalls;
    fl::flat_map<fl::string, PendingStream, fl::StringFastLess> mPendingStreams;
    fl::vector<fl::json> mIncomingQueue;
    int mNextCallId = 1;
 
    // Internal methods
    void sendHeartbeat();
    void checkHeartbeatTimeout(u32 currentTimeMs);
    bool processIncomingData();
    void handleConnectionStateChange(u32 currentTimeMs);
    void parseChunkedMessages();
    bool resolveRpc(const fl::json& msg, const fl::string& idKey);
    bool resolveRpcStream(const fl::json& msg, const fl::string& idKey);
    static fl::string idToString(const fl::json& id);
};
 
}  // namespace http
}  // namespace net
}  // namespace fl