auseft.Platform.DataSync/tcp_client/src/main.rs

mod db;

use std::net::TcpStream;
use std::io::{Read, Write};
use std::thread;
use std::time::Duration;
use serde::Deserialize;
use config::Config;
use db::{Database, DatabaseConfig, TableConfig};

#[derive(Debug, Deserialize)]
struct ServerConfig {
    host: String,
    port: u16,
}

#[derive(Debug, Deserialize)]
struct ClientConfig {
    max_retries: u32,
    retry_delay_secs: u64,
    read_timeout_secs: u64,
    write_timeout_secs: u64,
}

#[derive(Debug, Deserialize)]
struct SchedulerConfig {
    interval_hours: u64,
}

#[derive(Debug, Deserialize)]
struct Settings {
    server: ServerConfig,
    client: ClientConfig,
    database: DatabaseConfig,
    scheduler: SchedulerConfig,
    tables: Vec<TableConfig>,
}

fn load_config() -> Result<Settings, config::ConfigError> {
    let settings = Config::builder()
        .add_source(config::File::with_name("config"))
        .build()?;

    settings.try_deserialize()
}

async fn sync_data(db: &mut Database, config: &Settings) {
    for table in &config.tables {
        println!("\n开始获取表 {} 的数据...", table.name);
        match db.get_table_data(table).await {
            Ok(rows) => {
                println!("从数据库获取到 {} 条记录", rows.len());
                let mut success_count = 0;
                let mut fail_count = 0;

                // 将每行数据发送到服务器
                for (index, row) in rows.iter().enumerate() {
                    let msg = match serde_json::to_string(&row) {
                        Ok(msg) => msg,
                        Err(e) => {
                            println!("序列化数据失败 (第{}条记录): {}", index + 1, e);
                            fail_count += 1;
                            continue;
                        }
                    };

                    // 尝试发送数据到服务器
                    let mut retry_count = 0;
                    let server_addr = format!("{}:{}", config.server.host, config.server.port);
                    
                    while retry_count < config.client.max_retries {
                        match TcpStream::connect(&server_addr) {
                            Ok(mut stream) => {
                                // 设置读写超时
                                stream.set_read_timeout(Some(Duration::from_secs(config.client.read_timeout_secs))).unwrap_or_default();
                                stream.set_write_timeout(Some(Duration::from_secs(config.client.write_timeout_secs))).unwrap_or_default();

                                // 发送数据
                                if let Err(e) = stream.write_all(msg.as_bytes()) {
                                    println!("发送数据失败 (第{}条记录): {}", index + 1, e);
                                    retry_count += 1;
                                    thread::sleep(Duration::from_secs(config.client.retry_delay_secs));
                                    continue;
                                }

                                // 等待服务器响应
                                let mut response = [0; 1024];
                                match stream.read(&mut response) {
                                    Ok(_) => {
                                        success_count += 1;
                                        break;
                                    }
                                    Err(e) => {
                                        println!("读取服务器响应失败 (第{}条记录): {}", index + 1, e);
                                        retry_count += 1;
                                        thread::sleep(Duration::from_secs(config.client.retry_delay_secs));
                                        continue;
                                    }
                                }
                            }
                            Err(e) => {
                                println!("连接服务器失败 (第{}条记录): {}", index + 1, e);
                                retry_count += 1;
                                thread::sleep(Duration::from_secs(config.client.retry_delay_secs));
                                continue;
                            }
                        }
                    }

                    if retry_count >= config.client.max_retries {
                        println!("发送数据失败，已达到最大重试次数 (第{}条记录)", index + 1);
                        fail_count += 1;
                    }
                }

                println!("表 {} 同步完成", table.name);
                println!("成功: {} 条记录", success_count);
                println!("失败: {} 条记录", fail_count);
            }
            Err(e) => println!("获取表 {} 数据失败: {}", table.name, e),
        }
    }
}

#[tokio::main]
async fn main() {
    // 加载配置
    let config = match load_config() {
        Ok(cfg) => cfg,
        Err(e) => {
            println!("加载配置文件失败: {}", e);
            return;
        }
    };

    println!("配置信息:");
    println!("数据库主机: {}", config.database.host);
    println!("数据库端口: {}", config.database.port);
    println!("数据库名称: {}", config.database.name);
    println!("数据库用户: {}", config.database.user);
    println!("要同步的表数量: {}", config.tables.len());
    println!("定时任务间隔: {}小时", config.scheduler.interval_hours);

    loop {
        // 建立数据库连接
        let mut db = match Database::connect(&config.database).await {
            Ok(db) => {
                println!("数据库连接成功！");
                db
            }
            Err(e) => {
                println!("数据库连接失败: {}", e);
                println!("请检查数据库配置和网络连接:");
                println!("1. 确认数据库服务器是否运行");
                println!("2. 确认数据库服务器的IP地址是否正确");
                println!("3. 确认数据库服务器的端口是否正确");
                println!("4. 确认数据库用户和密码是否正确");
                println!("5. 确认网络连接是否正常");
                // 等待一段时间后重试
                tokio::time::sleep(Duration::from_secs(60)).await;
                continue;
            }
        };

        // 执行同步
        sync_data(&mut db, &config).await;

        // 等待下一次执行
        println!("\n等待 {} 小时后执行下一次同步...", config.scheduler.interval_hours);
        tokio::time::sleep(Duration::from_secs(config.scheduler.interval_hours * 3600)).await;
    }
}