Queue/readcard/Publisher.cxx

#include "Publisher.hpp"

#include <condition_variable>
#include <csignal>
#include <stdexcept>
#include <thread>

#include <sstream>
#include <iomanip>
#include <cstdio>

#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/log/Log.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
#include <fastdds/dds/publisher/qos/PublisherQos.hpp>

#include <fastdds/rtps/common/Types.hpp>
#include <fastdds/dds/core/Types.hpp>

#include "WeighingDDSTypePubSubTypes.hpp"
#include "msg.hpp"
#include "MsgHandler.hpp"
#include "DEBUG.hpp"

using namespace eprosima::fastdds::dds;

PublisherApp::PublisherApp(
        const int& domain_id)
    : factory_(nullptr)
    , participant_(nullptr)
    , publisher_(nullptr)
    , topic_(nullptr)
    , writer_(nullptr)
    , type_(new WeighingSystem::ReadCardRspPubSubType())
    , matched_(0)
    , samples_sent_(0)
    , stop_(false)
    , queue_trigger_publisher_(nullptr)
    , queue_trigger_topic_(nullptr)
    , queue_trigger_writer_(nullptr)
    , queue_trigger_type_(new Queue::QueueTriggerPubSubType())
    , current_date_()
    , sequence_num_(0)
    , last_card_()
    , current_message_id_()
    , wait_end_time_(std::chrono::steady_clock::now() - std::chrono::seconds(10))  // 初始化为过去时间，启动时不等待
    , last_trigger_card_()
    , wait_active_(false)                                  // 启动后不处于等待期，由 OnCardDetected 激活
{
    // Create the participant
    DomainParticipantQos pqos = PARTICIPANT_QOS_DEFAULT;
    pqos.name("ReadCard_pub_participant");
    pqos.wire_protocol().builtin.discovery_config.leaseDuration = Duration_t(60, 0);
    pqos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod = Duration_t(30, 0);
    factory_ = DomainParticipantFactory::get_shared_instance();
    participant_ = factory_->create_participant(domain_id, pqos, nullptr, StatusMask::none());
    if (participant_ == nullptr)
    {
        throw std::runtime_error("ReadCard Participant initialization failed");
    }

    // Register ReadCardRsp type
    type_.register_type(participant_);

    // Create the publisher (for ReadCardRsp)
    PublisherQos pub_qos = PUBLISHER_QOS_DEFAULT;
    participant_->get_default_publisher_qos(pub_qos);
    publisher_ = participant_->create_publisher(pub_qos, nullptr, StatusMask::none());
    if (publisher_ == nullptr)
    {
        throw std::runtime_error("ReadCard Publisher initialization failed");
    }

    // Create the topic (ReadCardRsp)
    TopicQos topic_qos = TOPIC_QOS_DEFAULT;
    participant_->get_default_topic_qos(topic_qos);
    topic_ = participant_->create_topic("ReadCardRsp", type_.get_type_name(), topic_qos);
    if (topic_ == nullptr)
    {
        throw std::runtime_error("ReadCard Topic initialization failed");
    }

    // Create the data writer (ReadCardRsp)
    DataWriterQos writer_qos = DATAWRITER_QOS_DEFAULT;
    publisher_->get_default_datawriter_qos(writer_qos);
    writer_qos.reliability().kind = ReliabilityQosPolicyKind::RELIABLE_RELIABILITY_QOS;
    writer_qos.durability().kind = DurabilityQosPolicyKind::VOLATILE_DURABILITY_QOS;
    writer_qos.history().kind = HistoryQosPolicyKind::KEEP_LAST_HISTORY_QOS;
    writer_ = publisher_->create_datawriter(topic_, writer_qos, this, StatusMask::all());
    if (writer_ == nullptr)
    {
        throw std::runtime_error("ReadCard DataWriter initialization failed");
    }

    // ── 创建 QueueTrigger 发布者 ──
    queue_trigger_type_.register_type(participant_);

    PublisherQos qt_pub_qos = PUBLISHER_QOS_DEFAULT;
    participant_->get_default_publisher_qos(qt_pub_qos);
    queue_trigger_publisher_ = participant_->create_publisher(qt_pub_qos, nullptr, StatusMask::none());

    TopicQos qt_topic_qos = TOPIC_QOS_DEFAULT;
    participant_->get_default_topic_qos(qt_topic_qos);
    queue_trigger_topic_ = participant_->create_topic(
            "QueueTrigger",
            queue_trigger_type_.get_type_name(),
            qt_topic_qos);

    DataWriterQos qt_writer_qos = DATAWRITER_QOS_DEFAULT;
    queue_trigger_publisher_->get_default_datawriter_qos(qt_writer_qos);
    qt_writer_qos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    qt_writer_qos.durability().kind = VOLATILE_DURABILITY_QOS;
    qt_writer_qos.history().kind = KEEP_LAST_HISTORY_QOS;
    queue_trigger_writer_ = queue_trigger_publisher_->create_datawriter(
            queue_trigger_topic_, qt_writer_qos, this, StatusMask::all());
}

PublisherApp::~PublisherApp()
{
    if (nullptr != participant_)
    {
        participant_->delete_contained_entities();
        factory_->delete_participant(participant_);
    }
}

void PublisherApp::on_publication_matched(
        DataWriter* writer,
        const PublicationMatchedStatus& info)
{
    if (info.current_count_change == 1)
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            matched_ = info.current_count;
        }
        DEBUG(writer->get_topic()->get_name() << " Publisher matched.");
        cv_.notify_one();
    }
    else if (info.current_count_change == -1)
    {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            matched_ = info.current_count;
        }
        DEBUG(writer->get_topic()->get_name() << " Publisher unmatched.");
    }
}

void PublisherApp::run(std::shared_ptr<MsgHandler> handler)
{
    // ── 拉卡检测回调：替代方案下放给主循环处理，这里不再绑定 ──
    // 真正的触发逻辑在 run() 主循环里：
    //   HandleDeviceMsg → 把卡号写入 handler->pending_trigger_card_
    //   主循环检测字段非空 → PublishQueueTrigger + PauseAutoFind + 启动 3s 等待期

    // Device->Core 的响应（DDS 载荷：index + map<string,sequence<octet>>）
    WeighingSystem::ReadCardRsp rsp;

    while (!is_stopped())
    {
        if (handler->isOpen() == true)
        {
            // ── 处理设备消息（FindCard 响应会更新 m_CardId 并写入 pending_trigger_card_） ──
            if (handler->HandleDeviceMsg(rsp))
            {
                // 当前版本不主动读卡，设备消息主要是 FindCard 响应
                rsp.msg().clear();
            }

            // ── 检测新卡：发 QueueTrigger + 启动 3s 等待期 ──
            std::string cardToTrigger;
            {
                std::lock_guard<std::mutex> lock(handler->trigger_mtx_);
                if (!handler->pending_trigger_card_.empty())
                {
                    cardToTrigger = handler->pending_trigger_card_;
                    handler->pending_trigger_card_.clear();  // 消费掉
                }
            }
            if (!cardToTrigger.empty())
            {
                // 防止同一张卡短时间内重复触发（3 秒等待期内忽略）
                auto now = std::chrono::steady_clock::now();
                if (now < wait_end_time_ && cardToTrigger == last_trigger_card_)
                {
                    DEBUG("Card " << cardToTrigger
                          << " still in 3s wait period, skipping QueueTrigger");
                }
                else
                {
                    PublishQueueTrigger(cardToTrigger);

                    // 暂停自动寻卡（state 切到 AUTO_FIND，m_AutoFind=false），
                    // 阻止 3 秒等待期内 HandleDdsMsg 接收 read/write
                    handler->PauseAutoFind();

                    wait_end_time_ = now + std::chrono::seconds(10);
                    wait_active_ = true;   // 标记等待期，主循环在 3s 后才调 ForceReset
                    last_trigger_card_ = cardToTrigger;
                    DEBUG("QueueTrigger sent for card=" << cardToTrigger
                          << ", waiting 10s before resuming auto-find");
                }
            }

            // ── 等待期检查 ──
            auto now2 = std::chrono::steady_clock::now();
            if (now2 >= wait_end_time_)
            {
                // 等待期刚结束：仅在这一次调用时恢复 AUTO_FIND
                if (wait_active_)
                {
                    static auto last_resume_log = std::chrono::steady_clock::time_point{};
                    if (last_resume_log.time_since_epoch().count() == 0
                            || now2 - last_resume_log > std::chrono::seconds(60))
                    {
                        DEBUG("10s wait elapsed, resuming AUTO_FIND");
                        last_resume_log = now2;
                    }
                    handler->ForceResetToAutoFind();
                    wait_active_ = false;
                }

                // pending read response（来自超时跳过）
                if (handler->HasPendingReadResponse())
                {
                    auto pendingRsp = handler->GetPendingReadResponse();
                    if (!pendingRsp.empty())
                    {
                        rsp.msg()["read"] = pendingRsp;
                        rsp.index(handler->GetCurrentIndex());
                        DEBUG("Publishing pending read response (from timeout skip)");
                        writer_->write(&rsp);
                        rsp.msg().clear();
                    }
                }
                handler->AutoFindTick();
            }
            else
            {
                // ── 等待期内：不发 DDS、不触发读卡、不恢复自动寻卡 ──
                auto remaining = std::chrono::duration_cast<std::chrono::milliseconds>(
                        wait_end_time_ - now2).count();
            }
        }

        std::unique_lock<std::mutex> period_lock(mutex_);
        cv_.wait_for(period_lock, std::chrono::milliseconds(10), [this]()
                {
                    return is_stopped();
                });
    }
}

bool PublisherApp::is_stopped()
{
    return stop_.load();
}

void PublisherApp::on_data_available(DataReader* /*reader*/)
{
    // 不再订阅任何 DDS Topic，无需处理
}

void PublisherApp::on_subscription_matched(
        DataReader* /*reader*/,
        const SubscriptionMatchedStatus& /*info*/)
{
    // 不再订阅任何 DDS Topic，无需处理
}

void PublisherApp::stop()
{
    stop_.store(true);
    cv_.notify_one();
}

std::string PublisherApp::GenerateMessageId(const std::string& card_number)
{
    std::lock_guard<std::mutex> lock(id_mtx_);

    time_t now = time(nullptr);
    char date_buf[9];
    strftime(date_buf, sizeof(date_buf), "%Y%m%d", localtime(&now));
    std::string today(date_buf);

    // 日期变了：重置序号，清空上次的 card 和 message_id
    if (today != current_date_)
    {
        current_date_ = today;
        sequence_num_ = 0;
        last_card_.clear();
        current_message_id_.clear();
    }

    // 同一天、相同卡号：复用上次的 message_id
    if (!last_card_.empty() && card_number == last_card_ && !current_message_id_.empty())
    {
        DEBUG("GenerateMessageId: card unchanged (" << card_number << "), reuse " << current_message_id_);
        return current_message_id_;
    }

    // 不同卡号：序号累加
    sequence_num_++;
    last_card_ = card_number;

    char id[32];
    snprintf(id, sizeof(id), "qm-%s-%06u", today.c_str(), sequence_num_);
    current_message_id_ = std::string(id);

    DEBUG("GenerateMessageId: new card=" << card_number
              << " -> " << current_message_id_);
    return current_message_id_;
}

void PublisherApp::PublishQueueTrigger(const std::string& card_number)
{
    Queue::QueueTrigger msg;
    const std::string mid = GenerateMessageId(card_number);
    msg.message_id(mid);
    msg.card_number(card_number);
    msg.device_id("3");   // 本机读卡器设备编号
    msg.trigger_time("");  // 暂时为 null（空字符串）
    queue_trigger_writer_->write(&msg);
    DEBUG("Published QueueTrigger: message_id=" << mid
          << " card=" << card_number << " device_id=3");
}