publishing.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <commrat/messaging/message_registry.hpp>
#include <commrat/mailbox/registry_mailbox.hpp>
#include <commrat/messages.hpp>  // TimsMessage definition
#include <commrat/module/helpers/address_helpers.hpp>  // encode_address, extract_*
#include <commrat/module/io/multi_output_manager.hpp>  // SubscriberInfo
#include <iostream>
#include <tuple>
#include <mutex>
#include <vector>
 
namespace commrat {
 
template<
    typename UserRegistry,
    typename OutputData,
    typename PublishMailboxT = TypedMailbox<UserRegistry>,
    typename ModuleType = void  // Module type for multi-output mailbox access
>
class Publisher {
protected:
    // References to module resources (set by derived class)
    // REMOVED: subscriber_manager_ - ALL modules now use MultiOutputManager (post-unification)
    // REMOVED: publish_mailbox_ - ALL modules now use MailboxSets (post-unification)
    ModuleType* module_ptr_{nullptr};  // Typed pointer to Module for mailbox/subscriber access
    std::string module_name_;
    
public:
    // REMOVED: set_subscriber_manager() - no longer used after unification
    // REMOVED: set_publish_mailbox() - no longer used after unification
    void set_module_ptr(ModuleType* ptr) { module_ptr_ = ptr; }
    void set_module_name(const std::string& name) { module_name_ = name; }
    
    template<typename T>
    static TimsMessage<T> create_tims_message(T&& payload, uint64_t timestamp_ns) {
        TimsMessage<T> msg{
            .header = {
                .msg_type = 0,     // serialize() will set this
                .msg_size = 0,     // serialize() will set this
                .timestamp = timestamp_ns,  // ONE SOURCE OF TRUTH
                .seq_number = 0,   // TiMS will set this
                .flags = 0
            },
            .payload = std::forward<T>(payload)
        };
        return msg;
    }
    
    template<typename T = OutputData>
        requires (!std::is_void_v<T>)
    void publish_to_subscribers(T& data) {
        if constexpr (!std::is_void_v<ModuleType>) {
            // Use CMD mailbox for publishing (Phase 7)
            auto& cmd_mbx = module_ptr_->template get_cmd_mailbox_public<0>();
            // Get output-specific subscriber list (index 0 for single-output)
            auto subscribers = module_ptr_->get_output_subscribers(0);
            
            for (const auto& sub : subscribers) {
                // Calculate destination: base_addr | mailbox_index
                uint32_t dest_mailbox = sub.base_addr | sub.input_index;
                auto result = cmd_mbx.send(data, dest_mailbox);
                if (!result) {
                    std::cerr << "[" << module_name_ << "] Send failed to subscriber base=0x" << std::hex << sub.base_addr 
                              << " mbx_idx=" << std::dec << static_cast<int>(sub.input_index)
                              << " dest=0x" << std::hex << dest_mailbox << std::dec
                              << " error=" << static_cast<int>(result.get_error()) << "\n";
                }
            }
        }
    }
    
    template<typename T>
    void publish_tims_message(TimsMessage<T>& tims_msg) {
        if constexpr (!std::is_void_v<ModuleType>) {
            // Use CMD mailbox for publishing (Phase 7)
            auto& cmd_mbx = module_ptr_->template get_cmd_mailbox_public<0>();
            // Get output-specific subscriber list (index 0 for single-output)
            auto subscribers = module_ptr_->get_output_subscribers(0);
            
            for (const auto& sub : subscribers) {
                uint32_t dest_mailbox = sub.base_addr | sub.input_index;
                // Phase 6.10: Send with explicit timestamp from header
                auto result = cmd_mbx.send(tims_msg.payload, dest_mailbox, tims_msg.header.timestamp);
                if (!result) {
                    std::cerr << "[" << module_name_ << "] Send failed to subscriber base=0x" << std::hex << sub.base_addr 
                              << " mbx_idx=" << std::dec << static_cast<int>(sub.input_index)
                              << " dest=0x" << std::hex << dest_mailbox << std::dec
                              << " error=" << static_cast<int>(result.get_error()) << "\n";
                }
            }
        }
    }
    
    template<std::size_t Index, typename OutputType>
    void send_output_at_index(uint16_t subscriber_type_id_low, OutputType& output, uint32_t dest_mailbox) {
        // Get the full message ID for this output type
        constexpr uint32_t output_msg_id = UserRegistry::template get_message_id<OutputType>();
        constexpr uint16_t output_type_id_low = static_cast<uint16_t>(output_msg_id & 0xFFFF);
        
        // Only send if types match
        if (output_type_id_low == subscriber_type_id_low) {
            // Get the correct publish mailbox for this output index (unified path)
            if constexpr (!std::is_void_v<ModuleType>) {
                auto& publish_mbx = module_ptr_->template get_publish_mailbox_public<Index>();
                auto result = publish_mbx.send(output, dest_mailbox);
                if (!result) {
                    std::cout << "[" << module_name_ << "] Send failed for output[" << Index << "]\n";
                }
            }
        }
    }
    
    template<typename... Ts, std::size_t... Is>
    void publish_multi_outputs_impl(std::tuple<Ts...>& outputs, std::index_sequence<Is...>) {
        // Each output index publishes to its own subscriber list
        (publish_output_at_index<Is>(std::get<Is>(outputs)), ...);
    }
    
    template<std::size_t Index, typename OutputType>
    void publish_output_at_index(OutputType& output) {
        if constexpr (!std::is_void_v<ModuleType>) {
            // Get subscribers for this specific output
            auto subscribers = module_ptr_->get_output_subscribers(Index);
            
            // Send to each subscriber using CMD mailbox
            auto& cmd_mbx = module_ptr_->template get_cmd_mailbox_public<Index>();
            for (const auto& sub : subscribers) {
                uint32_t dest_mailbox = sub.base_addr | sub.input_index;
                auto result = cmd_mbx.send(output, dest_mailbox);
                if (!result) {
                    std::cout << "[" << module_name_ << "] Send failed for output[" << Index 
                              << "] to subscriber base=0x" << std::hex << sub.base_addr 
                              << " mbx_idx=" << std::dec << static_cast<int>(sub.input_index)
                              << " dest=0x" << std::hex << dest_mailbox << std::dec << "\n";
                }
            }
        }
    }
    
    template<typename... Ts>
    void publish_multi_outputs(std::tuple<Ts...>& outputs) {
        publish_multi_outputs_impl(outputs, std::index_sequence_for<Ts...>{});
    }
    
    template<typename... Ts>
    void publish_multi_outputs_with_timestamp(std::tuple<Ts...>& outputs, uint64_t timestamp_ns) {
        // For multi-output, we still use the existing publish_multi_outputs
        // because each output goes through send_if_type_matches which wraps in TimsMessage
        // The timestamp is not used here yet - future enhancement could wrap each output
        publish_multi_outputs(outputs);
    }
};
 
} // namespace commrat