front_linux/LFtid1056/cloudfront/code/worker.cpp

////////////////////////////////////////////////////////////////////////////////////////////////////

#include <iostream>
#include <string>
#include <vector>
#include <array>
#include <list>
#include <map>
#include <thread>
#include <mutex>
#include <atomic>
#include <cstdio>
#include <sstream>
#include <algorithm>
#include <cctype>
#include <fnmatch.h>
#include <chrono>
#include <cstring>
#include <cstdlib>
#include <ctime>

// socket 网络编程
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <netdb.h>
#include <sys/select.h>      
#include <sys/stat.h>   

////////////////////////////////////////////////////////////////////////////////////////////////////

#include "worker.h"
#include "interface.h"
#include "rocketmq.h"
#include "log4cplus/log4.h"
#include "front.h"

/////////////////////////////////////////////////////////////////////////////////////////////////////

//shell日志打印开关
bool showinshellflag =false;

////////////////////////////////////////////////////////////////////////////////////////////////////

extern std::list<std::string> errorList, warnList, normalList;
extern std::mutex errorListMutex, warnListMutex, normalListMutex;

extern int IED_COUNT;
extern int INITFLAG;
extern int g_front_seg_index;
extern std::string subdir;

extern int G_TEST_NUM;
extern int G_TEST_TYPE;

extern bool errorOutputEnabled;
extern bool warnOutputEnabled;
extern bool normalOutputEnabled;

////////////////////////////////////////////////////////////////////////////////////////////////////测试登录类

    Worker::Worker(Front* front) 
        : m_front(front), listenFD(-1), running(false), historyIndex(-1), stopViewLog(true), g_stopTelnetTest(true) {}

    Worker::~Worker() {
        stopServer();
    }

    // 启动 Telnet 服务（监听端口）
    bool Worker::startServer(int port) {
        if (running) return false;

        listenFD = socket(AF_INET, SOCK_STREAM, 0);
        if (listenFD < 0) return false;

        int opt = 1;
        setsockopt(listenFD, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

        sockaddr_in addr{};
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = INADDR_ANY;
        addr.sin_port = htons(port);

        if (bind(listenFD, (sockaddr*)&addr, sizeof(addr)) < 0) return false;
        if (listen(listenFD, 5) < 0) return false;

        running = true;
        serverThread = std::thread(&Worker::acceptLoop, this);

        std::thread periodicThread(&Worker::doPeriodicTask, this);
        periodicThread.detach();

        return true;
    }

    // 停止 Telnet 服务
    void Worker::stopServer() {
        if (!running) return;
        running = false;
        shutdown(listenFD, SHUT_RDWR);
        close(listenFD);
        if (serverThread.joinable()) serverThread.join();
    }

    // 发送字符串
    void Worker::sendStr(int fd, const std::string& s) {
        send(fd, s.c_str(), s.size(), 0);
    }

    // 接收客户端连接主循环
    void Worker::acceptLoop() {
        while (running) {
            sockaddr_in caddr;
            socklen_t clen = sizeof(caddr);
            int cfd = accept(listenFD, (sockaddr*)&caddr, &clen);
            if (cfd >= 0) std::thread(&Worker::sessionThread, this, cfd).detach();
        }
    }

    //打印提示符
    void Worker::printPrompt(int clientFD)
    {
        sendStr(clientFD, "\r\n>");
    }

    // 客户端会话线程处理函数
    void Worker::sessionThread(int clientFD) {
        sendTelnetNegotiation(clientFD);
        sendStr(clientFD, "\r\nWelcome to the test shell. Type 'help'.\r\n> ");

        while (true) {
            char buf[512];
            int n = recv(clientFD, buf, sizeof(buf), 0);
            if (n <= 0) break;

            for (int i = 0; i < n; ++i) {
                unsigned char c = (unsigned char)buf[i];

                // 如果不是可打印字符，且也不是 上/下箭头、退格、回车等，就忽略
                bool isPrintable = (c >= 32 && c <= 126);   //可见字符范围
                bool isArrowOrEsc = (c == 0x1b);            //ESC
                bool isEnter = (c == '\r' || c == '\n');    //换行
                bool isBackspace = (c == 0x7f || c == 0x08);//退格
                bool isTab = (c == '\t');                   //tab
                if( !isPrintable && !isArrowOrEsc && !isEnter && !isBackspace && !isTab )
                {
                    // 跳过不认识的控制字符：如 '\0'
                    continue;
                }

                // Telnet 协议协商处理
                if (c == IAC) {
                    if (i + 1 < n && (buf[i+1] == TELDO || buf[i+1] == DONT || buf[i+1] == WILL || buf[i+1] == WONT)) i += 2;
                    else i++;
                    continue;
                }

                // 退出 viewlog / telnet 流程
                if (c == '`') {
                    stopViewLog = true;        // [MOD] 支持停止日志查看
                    g_stopTelnetTest = true;   // [MOD] 停止 telnet 测试
                    sendStr(clientFD, "\r\n[Log view stopped]\r\n> ");
                    continue;
                }

                // 2) 回车/换行 => 执行命令
                if (c == '\r' || c == '\n')
                {
                    // 如果当前是 '\r' 并且下一个是 '\n'，就跳过 '\n'
                    if (c == '\r') {
                        if (i + 1 < n && buf[i+1] == '\n') {
                            i++;
                        }
                        // 这里再加一步，如果紧跟着是 '\0'，也跳过
                        if (i + 1 < n && buf[i+1] == '\0') {
                            i++;
                        }
                    }
                    // 如果当前是 '\n'，也可以检查一下下一个是不是 '\0'（有些客户端可能发 '\n\0'）
                    else {
                        if (i + 1 < n && buf[i+1] == '\0') {
                            i++;
                        }
                    }
                
                    // 现在把 当前输入的指令 前后空白去掉
                    std::string cmdtrim = trim(currentCommand);
                    if (!cmdtrim.empty()) {     //输入指令非空则记录到历史
                        
                        if (commandHistory.empty() || commandHistory.back() != cmdtrim) {   //防止连续重复的历史记录
                            commandHistory.push_back(cmdtrim);
                        }

                        historyIndex = commandHistory.size();   //更新历史指令下标
                
                        processCommand(cmdtrim, clientFD);      //处理当前指令
                    }
                
                    // 处理后清空并打印新的提示符
                    currentCommand.clear();
                    printPrompt(clientFD);
                
                    // 回车后处理新的命令行
                    continue;
                }

                // 上/下箭头处理
                if (c == 0x1b && i + 2 < n && buf[i+1] == '[') {
                    char arrow = buf[i+2];
                    if (arrow == 'A') handleUpArrow(clientFD);
                    else if (arrow == 'B') handleDownArrow(clientFD);
                    i += 2;
                    continue;
                }

                // 退格处理
                if (c == 0x7f || c == 0x08) {
                    if (!currentCommand.empty()) {
                        currentCommand.pop_back();
                        sendStr(clientFD, "\b \b");
                    }
                    continue;
                }

                // 普通字符输入
                currentCommand.push_back((char)c);
                sendBytes(clientFD, (const char*)&c, 1);
            }
        }

        close(clientFD);
    }

    // 发送 Telnet 协商指令
    void Worker::sendTelnetNegotiation(int clientFD) {
        unsigned char will_echo[3] = { IAC, WILL, TELOPT_ECHO };
        unsigned char will_sga[3] = { IAC, WILL, TELOPT_SUPPRESS_GO_AHEAD };
        unsigned char dont_lin[3] = { IAC, DONT, TELOPT_LINEMODE };
        sendBytes(clientFD, (const char*)will_echo, 3);
        sendBytes(clientFD, (const char*)will_sga, 3);
        sendBytes(clientFD, (const char*)dont_lin, 3);
    }

    // 发送字节数组
    void Worker::sendBytes(int fd, const char* buf, int len) {
        send(fd, buf, len, 0);
    }

    void Worker::setTestNum(int num) {
        std::lock_guard<std::mutex> locker(testMutex);
        G_TEST_NUM = num;
    }

    void Worker::setTestType(int type) {
        std::lock_guard<std::mutex> locker(testMutex);
        G_TEST_TYPE = type;
    }

    // 日志控制
    void Worker::setTestlog(bool flag) {
        redirectErrorOutput(flag);
        redirectWarnOutput(flag);
        redirectNormalOutput(flag);
    }

    void Worker::doPeriodicTask() {
        while (running) {
            {
                std::lock_guard<std::mutex> locker(testMutex);
                std::cout << "[PeriodicTask] G_TEST_NUM = " << G_TEST_NUM 
                          << ", G_TEST_TYPE = " << G_TEST_TYPE << std::endl;

                if (G_TEST_NUM != 0) {
                    std::cout << "[PeriodicTask] Executing rocketmq_test_300()\n";
                    rocketmq_test_300(G_TEST_NUM, g_front_seg_index, G_TEST_TYPE,m_front);
                    //upload_data_test();
                }
            }

            std::this_thread::sleep_for(std::chrono::seconds(60)); // 每 60 秒执行一次
        }
    }

    // 命令处理逻辑扩展
    void Worker::processCommand(const std::string &cmd, int clientFD) {
        std::cout << "Received command: " << cmd << std::endl;

        if (cmd == "help") {
            std::string helpText =
                "Available commands:\r\n"
                "G_TEST_NUM=<num>   - Set the G_TEST_NUM\r\n"
                "G_TEST_TYPE=<num>   - Set the G_TEST_TYPE 0:use ledger,1:use number\r\n"
                "LOG=<bool>       - Set the LOG\r\n"
                "rc               - Execute rocketmq_test_rc\r\n"
                "rt               - Execute rocketmq_test_rt\r\n"
                "ud               - Execute rocketmq_test_ud\r\n"
                "set              - Execute rocketmq_test_set\r\n"
                "log              - Execute rocketmq_test_log\r\n"
                "upload           - Execute upload file\r\n"
                "qvvr             - Execute http_test_qvvr\r\n"
                "ledger <id>      - Execute ledger with optional terminal_id\r\n"
                "viewlog <level>  - View logs (ERROR, WARN, NORMAL, DEBUG)\r\n"
                "value <valuename> - Execute value print with valuename\r\n"
                "exit             - Exit the shell\r\n"
                "help             - Show this help message\r\n";
            sendStr(clientFD, "\r\x1B[K" + helpText);
        } else if (cmd.find("viewlog") == 0) {
            showinshellflag = true;
            handleViewLogCommand(cmd, clientFD);
        } else if (cmd.find("G_TEST_NUM=") == 0) {
            int num = std::atoi(cmd.substr(9).c_str());
            setTestNum(num);
            sendStr(clientFD, "\r\x1B[KTEST_NUM updated\r\n");
        } else if (cmd.find("G_TEST_TYPE=") == 0) {
            int type = std::atoi(cmd.substr(10).c_str());
            setTestType(type);
            sendStr(clientFD, "\r\x1B[KTEST_TYPE updated\r\n");
        } else if (cmd.find("LOG=") == 0) {
            int flag = std::atoi(cmd.substr(4).c_str());
            setTestlog(flag);
            sendStr(clientFD, "\r\x1B[KLOG updated\r\n");
        } else if (cmd == "rc") {
            rocketmq_test_rc(m_front);
            sendStr(clientFD, "\r\x1B[KExecuted rocketmq_test_rc\r\n");
        } else if (cmd == "rt") {
            rocketmq_test_rt(m_front);
            sendStr(clientFD, "\r\x1B[KExecuted rocketmq_test_rt\r\n");
        } else if (cmd == "ud") {
            rocketmq_test_ud(m_front);
            sendStr(clientFD, "\r\x1B[KExecuted rocketmq_test_ud\r\n");
        } else if (cmd == "set") {
            rocketmq_test_set(m_front);
            sendStr(clientFD, "\r\x1B[KExecuted rocketmq_test_set\r\n");
        } else if (cmd == "upload") {
            Fileupload_test();
            sendStr(clientFD, "\r\x1B[KExecuted upload file\r\n");
        } else if (cmd == "qvvr") {
            qvvr_test();
            sendStr(clientFD, "\r\x1B[KExecuted http_test_qvvr\r\n");
        } else if (cmd.find("ledger") == 0) {
            size_t pos = cmd.find(' ');
            if (pos != std::string::npos) {
                std::string terminalId = cmd.substr(pos + 1);
                ledger(terminalId.c_str(), clientFD);
                sendStr(clientFD, "\r\x1B[KExecuted ledger with terminal_id\r\n");
            } else {
                ledger("", clientFD);
                sendStr(clientFD, "\r\x1B[KExecuted ledger without parameters\r\n");
            }
        } else if (cmd.find("value") == 0) {
            size_t pos = cmd.find(' ');
            if (pos != std::string::npos) {
                std::string var = cmd.substr(pos + 1);
                sendStr(clientFD, "\r\x1B[KExecuted value with variable name: " + var + "\r\n");
                value_print(var, clientFD);
            } else {
                sendStr(clientFD, "\r\x1B[KPlease provide a variable name\r\n");
            }
        } else if (cmd == "exit") {
            sendStr(clientFD, "\r\x1B[KGoodbye! Exiting shell...\r\n");
            shutdown(clientFD, SHUT_RDWR);
            close(clientFD);
            return;
        } else {
            sendStr(clientFD, "\r\x1B[KUnknown command\r\n");
        }

        // 打印提示符
        sendStr(clientFD, "> ");
    }

    // 上箭头历史回溯
    void Worker::handleUpArrow(int fd) {
        if (!commandHistory.empty() && historyIndex > 0) {
            historyIndex--;
            currentCommand = commandHistory[historyIndex];
            sendStr(fd, "\r\x1B[K> " + currentCommand);
        }
    }

    // 下箭头历史前进
    void Worker::handleDownArrow(int fd) {
        if (!commandHistory.empty() && historyIndex < (int)commandHistory.size() - 1) {
            historyIndex++;
            currentCommand = commandHistory[historyIndex];
            sendStr(fd, "\r\x1B[K> " + currentCommand);
        } else {
            currentCommand.clear();
            sendStr(fd, "\r\x1B[K> ");
        }
    }

    // 字符串 trim
    std::string Worker::trim(const std::string& s) {
        auto start = s.begin();
        while (start != s.end() && std::isspace(*start)) ++start;
        auto end = s.end();
        do { --end; } while (std::distance(start, end) > 0 && std::isspace(*end));
        return (start <= end) ? std::string(start, end + 1) : "";
    }

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////测试shell用的函数

void Worker::printLedgerinshell(const terminal_dev& dev, int fd) {
    std::ostringstream os;
    os << "\r\x1B[K------------------------------------\n";
    os << "\r\x1B[K|-- terminal_id: " << dev.terminal_id << "\n";
    os << "\r\x1B[K|-- terminal_name: " << dev.terminal_name << "\n";
    os << "\r\x1B[K|-- dev_ip: " << dev.addr_str << "\n";
    os << "\r\x1B[K|-- dev_port: " << dev.port << "\n";
    os << "\r\x1B[K|-- dev_type: " << dev.dev_type << "\n";
    os << "\r\x1B[K|-- dev_key: " << dev.dev_key << "\n";
    os << "\r\x1B[K|-- dev_series: " << dev.dev_series << "\n";
    os << "\r\x1B[K|-- dev_processNo: " << dev.processNo << "\n";
    os << "\r\x1B[K|-- maxProcessNum: " << dev.maxProcessNum << "\n";
    os << "\r\x1B[K|-- org_name: " << dev.org_name << "\n";
    os << "\r\x1B[K|-- maint_name: " << dev.maint_name << "\n";
    os << "\r\x1B[K|-- station_name: " << dev.station_name << "\n";
    os << "\r\x1B[K|-- tmnl_factory: " << dev.tmnl_factory << "\n";
    os << "\r\x1B[K|-- tmnl_status: " << dev.tmnl_status << "\n";
    os << "\r\x1B[K|-- timestamp: " << dev.timestamp << "\n";

    os << "\r\x1B[K|-- mac: " << dev.mac << "\n";

    for (size_t i = 0; i < dev.line.size(); ++i) {
        const auto& ld = dev.line[i];
        if (ld.monitor_id.empty()) continue;
        os << "\r\x1B[K|-- line[" << i << "]:\n";
        os << "\r\x1B[K    |-- monitor_id: " << ld.monitor_id << "\n";
        os << "\r\x1B[K    |-- monitor_name: " << ld.monitor_name << "\n";
        os << "\r\x1B[K    |-- logical_device_seq: " << ld.logical_device_seq << "\n";
        os << "\r\x1B[K    |-- terminal_id: " << ld.terminal_id << "\n";
        os << "\r\x1B[K    |-- voltage_level: " << ld.voltage_level << "\n";
        os << "\r\x1B[K    |-- terminal_connect: " << ld.terminal_connect << "\n";
        os << "\r\x1B[K    |-- status: " << ld.status << "\n";
        os << "\r\x1B[K    |-- timestamp: " << ld.timestamp << "\n";

        os << "\r\x1B[K    |-- CT1: " << ld.CT1 << "\n";
        os << "\r\x1B[K    |-- CT2: " << ld.CT2 << "\n";
        os << "\r\x1B[K    |-- PT1: " << ld.PT1 << "\n";
        os << "\r\x1B[K    |-- PT2: " << ld.PT2 << "\n";

    }

    os << "\r\x1B[K------------------------------------\n";
    sendStr(fd, os.str());
}

void Worker::ledger(const std::string& terminal_id, int fd) {
    sendStr(fd, "\r\x1B[Kprint ledger in shell\n");
    std::lock_guard<std::mutex> lock(ledgermtx);
    bool found = false;

    if (terminal_id.empty()) {
        for (const auto& dev : terminal_devlist) {
            printLedgerinshell(dev, fd);
        }
    } else {
        for (const auto& dev : terminal_devlist) {
            if (dev.terminal_id == terminal_id) {
                printLedgerinshell(dev, fd);
                found = true;
                break;
            }
        }

        if (!found) {
            std::ostringstream msg;
            msg << "\r\x1B[Kterminal not exist: " << terminal_id << "\n";
            sendStr(fd, msg.str());
        }
    }
}

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////打印指定的变量名

void Worker::value_print(const std::string& variableName, int clientFD) {
    std::string output;

    {
        std::lock_guard<std::mutex> lock(ledgermtx);
        std::cout << "value_print hold lock !!!!!!!!!!!" << std::endl;

        if (variableName == "frontindex") {
            output = "frontindex = " + std::to_string(g_front_seg_index);
        } else if (variableName == "iedcount") {
            output = "ledger list = " + std::to_string(terminal_devlist.size()) +
                     ", ied config count = " + std::to_string(IED_COUNT);
        } else if (variableName == "frontfun") {
            output = "frontfun = " + subdir;
        } else if (variableName == "log") {
            output = "showinshellflag = " + std::to_string(showinshellflag) +
                     ", normalOutputEnabled = " + std::to_string(normalOutputEnabled) +
                     ", warnOutputEnabled = " + std::to_string(warnOutputEnabled) +
                     ", errorOutputEnabled = " + std::to_string(errorOutputEnabled);
        } else if (variableName == "init") {
            output = "INITFLAG = " + std::to_string(INITFLAG);
        } else {
            output = "Unknown variable name: " + variableName;
        }

        std::cout << "value_print free lock !!!!!!!!!!!" << std::endl;
    }

    sendStr(clientFD, "\r\x1B[K" + output + "\r\n");
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////////日志开关控制

//日志消息列表
std::list<std::string>* getLogList(const std::string& level) {
    if (level == "ERROR") return &errorList;
    if (level == "WARN") return &warnList;
    if (level == "NORMAL") return &normalList;
    return nullptr;
}

//日志锁
std::mutex* getLogMutex(const std::string& level) {
    if (level == "ERROR") return &errorListMutex;
    if (level == "WARN") return &warnListMutex;
    if (level == "NORMAL") return &normalListMutex;
    return nullptr;
}

void Worker::handleViewLogCommand(const std::string& command, int clientFD) {
    std::istringstream iss(command);
    std::string cmd, level;
    iss >> cmd >> level;
    std::transform(level.begin(), level.end(), level.begin(), ::toupper);

    if (level.empty()) {
        sendStr(clientFD, "\r\x1B[KUsage: viewlog [ERROR|WARN|NORMAL]\r\n> ");
        return;
    }

    std::list<std::string>* logList = getLogList(level);
    std::mutex* logMutex = getLogMutex(level);

    if (!logList || !logMutex) {
        sendStr(clientFD, "\r\x1B[KInvalid log level! Use ERROR, WARN, NORMAL.\r\n> ");
        return;
    }

    stopViewLog = false;
    showinshellflag = true;
    sendStr(clientFD, "\r\x1B[KViewing logs for level: " + level + " (Press '`' to exit)\r\n> ");

    char inputBuf[16];

    while (!stopViewLog) {
        // 1. 监听 shell 输入退出符号 `
        fd_set read_fds;
        FD_ZERO(&read_fds);
        FD_SET(clientFD, &read_fds);

        timeval timeout;
        timeout.tv_sec = 0;
        timeout.tv_usec = 500000; // 500ms

        int activity = select(clientFD + 1, &read_fds, nullptr, nullptr, &timeout);
        if (activity > 0 && FD_ISSET(clientFD, &read_fds)) {
            int n = recv(clientFD, inputBuf, sizeof(inputBuf), 0);
            if (n > 0 && strchr(inputBuf, '`')) {
                stopViewLog = true;
                showinshellflag = false;
                break;
            }
        }

        // 2. 输出日志
        std::string logEntry;
        {
            std::lock_guard<std::mutex> lock(*logMutex);
            if (!logList->empty()) {
                logEntry = logList->front();
                logList->pop_front();
            }
        }

        if (!logEntry.empty()) {
            sendStr(clientFD, "\r\x1B[K" + logEntry + "\r\n");
        } else {
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    }

    // 3. 打印退出提示
    sendStr(clientFD, "\r\x1B[K\nLog view stopped. Returning to shell.\r\n> ");
}