message_channel.hpp

Go to the documentation of this file.

#pragma once
 
// MessageChannel — structured-message layer on top of byte streams.
//
// `Stream` and `AsyncStream` deal in raw bytes. MessageChannel deals in
// whole messages: one `send()` transmits one message, one `on_message`
// callback fires per received message. This is the right abstraction for
// WebSocket, OSC, JSON-RPC, and similar request/response protocols where
// a partial read is always meaningless.
//
// Layering:
//   byte transport  →  Stream / AsyncStream
//   framing         →  MessageChannel implementations
//   protocol        →  JsonRpcClient, remote-control, user code
//
// All MessageChannel methods are thread-safe unless documented otherwise.
// Callback dispatch follows the same executor contract as AsyncStream: if
// a `MessageExecutor` is supplied at construction, callbacks land on that
// executor; otherwise they run on whichever thread the implementation uses
// internally (usually its reader thread).
 
#include <cstddef>
#include <cstdint>
#include <functional>
#include <string>
#include <string_view>
#include <vector>
 
namespace pulp::runtime {

enum class MessageKind : uint8_t {
    Binary,  
    Text,    
};

struct Message {
    MessageKind kind = MessageKind::Binary;
    std::vector<std::uint8_t> payload;

    std::string_view as_text() const {
        return {reinterpret_cast<const char*>(payload.data()), payload.size()};
    }
};

using MessageCallback = std::function<void(const Message& message)>;

using ChannelClosedCallback = std::function<void()>;

using ChannelErrorCallback = std::function<void(std::string_view reason)>;

using MessageExecutor = std::function<void(std::function<void()>)>;

class MessageChannel {
public:
    virtual ~MessageChannel() = default;
 
    MessageChannel(const MessageChannel&) = delete;
    MessageChannel& operator=(const MessageChannel&) = delete;

    virtual bool send(const std::uint8_t* data, std::size_t size) = 0;
 
    bool send(std::string_view text) {
        return send_text(text);
    }
 
    virtual bool send_text(std::string_view text) {
        // Default: transports that don't distinguish just send bytes.
        return send(reinterpret_cast<const std::uint8_t*>(text.data()), text.size());
    }

    virtual void on_message(MessageCallback callback) = 0;
    virtual void on_closed(ChannelClosedCallback callback) = 0;
    virtual void on_error(ChannelErrorCallback callback) = 0;

    virtual void close() = 0;

    virtual bool is_open() const = 0;
 
protected:
    MessageChannel() = default;
};
 
}  // namespace pulp::runtime