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

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use config::Config;
use tokio_postgres::{NoTls, Error as PgError};
use serde::{Deserialize, Serialize};
use std::sync::Arc;

#[derive(Debug, Serialize, Deserialize)]
struct InstrumentInfo {
    id: i32,
    informationid: i32,
    instrumentcode: String,
    laboratoryid: i32,
    name: String,
    remark: String,
    specification: String,
}

async fn connect_db(config: &Config) -> Result<tokio_postgres::Client, PgError> {
    let host = config.get_string("database.host").unwrap();
    let port = config.get_int("database.port").unwrap() as u16;
    let dbname = config.get_string("database.name").unwrap();
    let user = config.get_string("database.user").unwrap();
    let password = config.get_string("database.password").unwrap();

    let connection_string = format!(
        "host={} port={} dbname={} user={} password={}",
        host, port, dbname, user, password
    );

    let (client, connection) = tokio_postgres::connect(&connection_string, NoTls).await?;
    
    // 在后台运行连接
    tokio::spawn(async move {
        if let Err(e) = connection.await {
            eprintln!("数据库连接错误: {}", e);
        }
    });

    Ok(client)
}

async fn insert_instrument(client: &tokio_postgres::Client, info: &InstrumentInfo) -> Result<(), PgError> {
    // 先检查 ID 是否存在
    let exists = client
        .query_one(
            "SELECT EXISTS(SELECT 1 FROM public.hy_instrument WHERE id = $1)",
            &[&info.id],
        )
        .await?
        .get::<_, bool>(0);

    if exists {
        println!("ID {} 已存在，跳过插入", info.id);
        return Ok(());
    }

    // ID 不存在，执行插入
    client.execute(
        "INSERT INTO public.hy_instrument (id, informationid, instrumentcode, laboratoryid, name, remark, specification) 
         OVERRIDING SYSTEM VALUE
         VALUES ($1, $2, $3, $4, $5, $6, $7)",
        &[
            &info.id,
            &info.informationid,
            &info.instrumentcode,
            &info.laboratoryid,
            &info.name,
            &info.remark,
            &info.specification,
        ],
    )
    .await?;

    println!("成功插入仪器信息: {} (ID: {})", info.instrumentcode, info.id);
    Ok(())
}

fn check_ok_message(message: &[u8]) -> u8 {
    0xFF
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 读取配置文件
    let settings = Config::builder()
        .add_source(config::File::with_name("config"))
        .build()?;

    // 连接数据库
    let client = connect_db(&settings).await?;
    let client = Arc::new(client);
    println!("数据库连接成功");

    let address = settings.get_string("server.address")?;
    let port = settings.get_int("server.port")? as u16;
    let bind_address = format!("{}:{}", address, port);

    let listener = TcpListener::bind(&bind_address).await?;
    println!("服务器监听地址: {}", bind_address);

    loop {
        let (mut socket, addr) = listener.accept().await?;
        println!("新客户端连接: {}", addr);
        let client = Arc::clone(&client);

        tokio::spawn(async move {
            let mut buf = [0; 1024 * 64];  // 增加缓冲区大小到64KB

            loop {
                match socket.read(&mut buf).await {
                    Ok(0) => {
                        println!("客户端断开连接: {}", addr);
                        return;
                    }
                    Ok(n) => {
                        let data = &buf[..n];
                        match serde_json::from_slice::<InstrumentInfo>(data) {
                            Ok(info) => {
                                println!("接收到仪器信息: {:?}", info);
                                if let Err(e) = insert_instrument(&client, &info).await {
                                    eprintln!("插入数据失败: {}", e);
                                }
                            }
                            Err(e) => {
                                eprintln!("解析JSON失败: {}", e);
                                eprintln!("接收到的数据: {}", String::from_utf8_lossy(data));
                            }
                        }

                        let response = check_ok_message(data);
                        if let Err(e) = socket.write_all(&[response]).await {
                            println!("发送响应失败: {}", e);
                            return;
                        }
                    }
                    Err(e) => {
                        println!("读取数据失败: {}", e);
                        return;
                    }
                }
            }
        });
    }
}