// TCP：面向连接，可靠
// 优点：可靠、有序、有流量控制
// 缺点：开销大、速度相对慢

// UDP：无连接，不可靠
// 优点：速度快、开销小
// 缺点：不可靠、无序、无流量控制

// 三次握手过程
// 1. 客户端发送SYN
// 2. 服务端响应SYN-ACK
// 3. 客户端发送ACK

public class TCPHandshake {
    
    // 客户端
    public void connect() {
        // 1. 发送SYN
        sendSYN();
        
        // 2. 接收SYN-ACK
        receiveSYNACK();
        
        // 3. 发送ACK
        sendACK();
        
        // 连接建立成功
    }
    
    // 服务端
    public void accept() {
        // 1. 接收SYN
        receiveSYN();
        
        // 2. 发送SYN-ACK
        sendSYNACK();
        
        // 3. 接收ACK
        receiveACK();
        
        // 连接建立成功
    }
}

import java.io.*;
import java.net.*;
import java.util.concurrent.*;

public class IoTTCPServer {
    
    private ServerSocket serverSocket;
    private ExecutorService executorService;
    private Map<String, Socket> deviceConnections;
    
    public IoTTCPServer(int port) throws IOException {
        this.serverSocket = new ServerSocket(port);
        this.executorService = Executors.newCachedThreadPool();
        this.deviceConnections = new ConcurrentHashMap<>();
    }
    
    public void start() {
        System.out.println("IoT TCP Server started on port: " + serverSocket.getLocalPort());
        
        while (true) {
            try {
                // 接受客户端连接
                Socket clientSocket = serverSocket.accept();
                String deviceId = getDeviceId(clientSocket);
                
                System.out.println("Device connected: " + deviceId);
                
                // 保存设备连接
                deviceConnections.put(deviceId, clientSocket);
                
                // 为每个设备创建处理线程
                executorService.submit(new DeviceHandler(clientSocket, deviceId));
                
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    private String getDeviceId(Socket socket) {
        // 从连接中获取设备ID
        return socket.getRemoteSocketAddress().toString();
    }
    
    // 设备处理器
    private class DeviceHandler implements Runnable {
        private Socket socket;
        private String deviceId;
        private BufferedReader reader;
        private PrintWriter writer;
        
        public DeviceHandler(Socket socket, String deviceId) {
            this.socket = socket;
            this.deviceId = deviceId;
        }
        
        @Override
        public void run() {
            try {
                reader = new BufferedReader(
                    new InputStreamReader(socket.getInputStream(), "UTF-8")
                );
                writer = new PrintWriter(
                    new OutputStreamWriter(socket.getOutputStream(), "UTF-8"),
                    true
                );
                
                String message;
                while ((message = reader.readLine()) != null) {
                    // 处理设备消息
                    handleDeviceMessage(deviceId, message);
                }
                
            } catch (IOException e) {
                System.out.println("Device disconnected: " + deviceId);
            } finally {
                cleanup();
            }
        }
        
        private void handleDeviceMessage(String deviceId, String message) {
            System.out.println("Received from " + deviceId + ": " + message);
            // 处理业务逻辑
        }
        
        private void cleanup() {
            try {
                if (reader != null) reader.close();
                if (writer != null) writer.close();
                if (socket != null) socket.close();
                deviceConnections.remove(deviceId);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public void sendToDevice(String deviceId, String message) {
        Socket socket = deviceConnections.get(deviceId);
        if (socket != null && !socket.isClosed()) {
            try {
                PrintWriter writer = new PrintWriter(
                    new OutputStreamWriter(socket.getOutputStream(), "UTF-8"),
                    true
                );
                writer.println(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

import java.io.*;
import java.net.*;

public class IoTTCPClient {
    
    private Socket socket;
    private BufferedReader reader;
    private PrintWriter writer;
    private String serverHost;
    private int serverPort;
    private String deviceId;
    private volatile boolean connected = false;
    
    public IoTTCPClient(String serverHost, int serverPort, String deviceId) {
        this.serverHost = serverHost;
        this.serverPort = serverPort;
        this.deviceId = deviceId;
    }
    
    public boolean connect() {
        try {
            socket = new Socket(serverHost, serverPort);
            socket.setKeepAlive(true);
            socket.setSoTimeout(30000);  // 30秒超时
            
            reader = new BufferedReader(
                new InputStreamReader(socket.getInputStream(), "UTF-8")
            );
            writer = new PrintWriter(
                new OutputStreamWriter(socket.getOutputStream(), "UTF-8"),
                true
            );
            
            // 发送设备ID
            sendMessage("DEVICE_ID:" + deviceId);
            
            connected = true;
            System.out.println("Connected to server: " + serverHost + ":" + serverPort);
            
            // 启动接收线程
            new Thread(this::receiveMessages).start();
            
            return true;
            
        } catch (IOException e) {
            System.out.println("Failed to connect: " + e.getMessage());
            connected = false;
            return false;
        }
    }
    
    public void sendMessage(String message) {
        if (connected && writer != null) {
            writer.println(message);
        }
    }
    
    private void receiveMessages() {
        try {
            String message;
            while (connected && (message = reader.readLine()) != null) {
                handleServerMessage(message);
            }
        } catch (IOException e) {
            System.out.println("Connection lost: " + e.getMessage());
            connected = false;
        }
    }
    
    private void handleServerMessage(String message) {
        System.out.println("Received from server: " + message);
        // 处理服务器消息
    }
    
    public void disconnect() {
        connected = false;
        try {
            if (reader != null) reader.close();
            if (writer != null) writer.close();
            if (socket != null) socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    public boolean isConnected() {
        return connected && socket != null && !socket.isClosed();
    }
}

// 消息格式：消息类型|消息长度|消息体
// 例如：DATA|100|{...}

public class IoTMessage {
    
    public enum MessageType {
        HEARTBEAT,      // 心跳
        DATA,           // 数据
        COMMAND,        // 命令
        RESPONSE,       // 响应
        ERROR           // 错误
    }
    
    private MessageType type;
    private int length;
    private String body;
    private long timestamp;
    
    public IoTMessage(MessageType type, String body) {
        this.type = type;
        this.body = body;
        this.length = body.getBytes().length;
        this.timestamp = System.currentTimeMillis();
    }
    
    // 编码
    public String encode() {
        return type.name() + "|" + length + "|" + body;
    }
    
    // 解码
    public static IoTMessage decode(String message) {
        String[] parts = message.split("\\|", 3);
        if (parts.length != 3) {
            throw new IllegalArgumentException("Invalid message format");
        }
        
        MessageType type = MessageType.valueOf(parts[0]);
        int length = Integer.parseInt(parts[1]);
        String body = parts[2];
        
        IoTMessage msg = new IoTMessage(type, body);
        return msg;
    }
    
    // Getters
    public MessageType getType() { return type; }
    public int getLength() { return length; }
    public String getBody() { return body; }
    public long getTimestamp() { return timestamp; }
}

public class DataSender {
    
    private IoTTCPClient client;
    private BlockingQueue<IoTMessage> messageQueue;
    
    public DataSender(IoTTCPClient client) {
        this.client = client;
        this.messageQueue = new LinkedBlockingQueue<>();
    }
    
    public void sendData(String data) {
        IoTMessage message = new IoTMessage(
            IoTMessage.MessageType.DATA,
            data
        );
        messageQueue.offer(message);
    }
    
    public void start() {
        new Thread(() -> {
            while (client.isConnected()) {
                try {
                    IoTMessage message = messageQueue.take();
                    String encoded = message.encode();
                    client.sendMessage(encoded);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
}

import java.util.concurrent.*;

public class HeartbeatManager {
    
    private IoTTCPClient client;
    private ScheduledExecutorService scheduler;
    private long heartbeatInterval = 30;  // 30秒
    private long timeout = 90;  // 90秒超时
    private long lastHeartbeatTime;
    private volatile boolean running = false;
    
    public HeartbeatManager(IoTTCPClient client) {
        this.client = client;
        this.scheduler = Executors.newScheduledThreadPool(2);
        this.lastHeartbeatTime = System.currentTimeMillis();
    }
    
    public void start() {
        running = true;
        
        // 定时发送心跳
        scheduler.scheduleAtFixedRate(
            this::sendHeartbeat,
            0,
            heartbeatInterval,
            TimeUnit.SECONDS
        );
        
        // 定时检查超时
        scheduler.scheduleAtFixedRate(
            this::checkTimeout,
            timeout,
            timeout,
            TimeUnit.SECONDS
        );
    }
    
    private void sendHeartbeat() {
        if (client.isConnected() && running) {
            IoTMessage heartbeat = new IoTMessage(
                IoTMessage.MessageType.HEARTBEAT,
                "PING"
            );
            client.sendMessage(heartbeat.encode());
        }
    }
    
    public void receiveHeartbeat() {
        lastHeartbeatTime = System.currentTimeMillis();
    }
    
    private void checkTimeout() {
        long elapsed = System.currentTimeMillis() - lastHeartbeatTime;
        if (elapsed > timeout * 1000) {
            System.out.println("Heartbeat timeout, connection may be lost");
            // 触发重连
            client.disconnect();
        }
    }
    
    public void stop() {
        running = false;
        scheduler.shutdown();
    }
}

public class ServerHeartbeatHandler {
    
    private Map<String, Long> deviceLastHeartbeat;
    
    public ServerHeartbeatHandler() {
        this.deviceLastHeartbeat = new ConcurrentHashMap<>();
    }
    
    public void handleHeartbeat(String deviceId, IoTMessage message) {
        if (message.getType() == IoTMessage.MessageType.HEARTBEAT) {
            // 更新最后心跳时间
            deviceLastHeartbeat.put(deviceId, System.currentTimeMillis());
            
            // 响应心跳
            IoTMessage response = new IoTMessage(
                IoTMessage.MessageType.HEARTBEAT,
                "PONG"
            );
            // 发送响应
        }
    }
    
    public void checkDeviceHealth() {
        long currentTime = System.currentTimeMillis();
        long timeout = 120000;  // 2分钟超时
        
        deviceLastHeartbeat.entrySet().removeIf(entry -> {
            if (currentTime - entry.getValue() > timeout) {
                System.out.println("Device timeout: " + entry.getKey());
                // 清理设备连接
                return true;
            }
            return false;
        });
    }
}

import java.util.concurrent.*;

public class ReconnectionManager {
    
    private IoTTCPClient client;
    private ScheduledExecutorService scheduler;
    private int maxRetries = 10;
    private long initialDelay = 1;  // 1秒
    private long maxDelay = 60;     // 60秒
    private int currentRetry = 0;
    
    public ReconnectionManager(IoTTCPClient client) {
        this.client = client;
        this.scheduler = Executors.newScheduledThreadPool(1);
    }
    
    public void startReconnect() {
        if (currentRetry >= maxRetries) {
            System.out.println("Max retries reached, stop reconnecting");
            return;
        }
        
        // 指数退避
        long delay = Math.min(
            initialDelay * (long) Math.pow(2, currentRetry),
            maxDelay
        );
        
        System.out.println("Reconnecting in " + delay + " seconds... (attempt " + (currentRetry + 1) + ")");
        
        scheduler.schedule(() -> {
            if (!client.isConnected()) {
                boolean success = client.connect();
                if (success) {
                    System.out.println("Reconnected successfully");
                    currentRetry = 0;
                } else {
                    currentRetry++;
                    startReconnect();
                }
            }
        }, delay, TimeUnit.SECONDS);
    }
    
    public void reset() {
        currentRetry = 0;
    }
}

public class ConnectionMonitor {
    
    private IoTTCPClient client;
    private ScheduledExecutorService scheduler;
    
    public ConnectionMonitor(IoTTCPClient client) {
        this.client = client;
        this.scheduler = Executors.newScheduledThreadPool(1);
    }
    
    public void start() {
        scheduler.scheduleAtFixedRate(
            this::checkConnection,
            10,
            10,
            TimeUnit.SECONDS
        );
    }
    
    private void checkConnection() {
        if (!client.isConnected()) {
            System.out.println("Connection lost, triggering reconnect");
            // 触发重连
        }
    }
    
    public void stop() {
        scheduler.shutdown();
    }
}

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

public class MessageCodec {
    
    private Gson gson;
    
    public MessageCodec() {
        this.gson = new GsonBuilder()
            .setDateFormat("yyyy-MM-dd HH:mm:ss")
            .create();
    }
    
    // 编码
    public String encode(Object obj) {
        return gson.toJson(obj);
    }
    
    // 解码
    public <T> T decode(String json, Class<T> clazz) {
        return gson.fromJson(json, clazz);
    }
}

// 设备数据模型
public class DeviceData {
    private String deviceId;
    private String sensorType;
    private double value;
    private long timestamp;
    
    // Constructors, getters, setters
    public DeviceData(String deviceId, String sensorType, double value) {
        this.deviceId = deviceId;
        this.sensorType = sensorType;
        this.value = value;
        this.timestamp = System.currentTimeMillis();
    }
    
    // Getters and setters
    public String getDeviceId() { return deviceId; }
    public void setDeviceId(String deviceId) { this.deviceId = deviceId; }
    public String getSensorType() { return sensorType; }
    public void setSensorType(String sensorType) { this.sensorType = sensorType; }
    public double getValue() { return value; }
    public void setValue(double value) { this.value = value; }
    public long getTimestamp() { return timestamp; }
    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
}

import java.io.*;
import java.nio.ByteBuffer;

public class BinaryCodec {
    
    // 编码
    public byte[] encode(IoTMessage message) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        
        try {
            // 消息类型（1字节）
            dos.writeByte(message.getType().ordinal());
            
            // 时间戳（8字节）
            dos.writeLong(message.getTimestamp());
            
            // 消息体长度（4字节）
            byte[] bodyBytes = message.getBody().getBytes("UTF-8");
            dos.writeInt(bodyBytes.length);
            
            // 消息体
            dos.write(bodyBytes);
            
            return baos.toByteArray();
            
        } catch (IOException e) {
            throw new RuntimeException("Encode error", e);
        }
    }
    
    // 解码
    public IoTMessage decode(byte[] data) {
        ByteArrayInputStream bais = new ByteArrayInputStream(data);
        DataInputStream dis = new DataInputStream(bais);
        
        try {
            // 消息类型
            int typeOrdinal = dis.readByte();
            IoTMessage.MessageType type = IoTMessage.MessageType.values()[typeOrdinal];
            
            // 时间戳
            long timestamp = dis.readLong();
            
            // 消息体长度
            int bodyLength = dis.readInt();
            
            // 消息体
            byte[] bodyBytes = new byte[bodyLength];
            dis.readFully(bodyBytes);
            String body = new String(bodyBytes, "UTF-8");
            
            IoTMessage message = new IoTMessage(type, body);
            return message;
            
        } catch (IOException e) {
            throw new RuntimeException("Decode error", e);
        }
    }
}

// 主程序：服务端
public class IoTTCPServerMain {
    
    public static void main(String[] args) {
        try {
            IoTTCPServer server = new IoTTCPServer(8888);
            server.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

// 主程序：客户端
public class IoTTCPClientMain {
    
    public static void main(String[] args) {
        String deviceId = "DEVICE_001";
        IoTTCPClient client = new IoTTCPClient("localhost", 8888, deviceId);
        
        // 连接
        if (client.connect()) {
            // 启动心跳
            HeartbeatManager heartbeat = new HeartbeatManager(client);
            heartbeat.start();
            
            // 启动数据发送
            DataSender sender = new DataSender(client);
            sender.start();
            
            // 模拟发送数据
            MessageCodec codec = new MessageCodec();
            for (int i = 0; i < 10; i++) {
                DeviceData data = new DeviceData(
                    deviceId,
                    "TEMPERATURE",
                    25.5 + i
                );
                String json = codec.encode(data);
                sender.sendData(json);
                
                Thread.sleep(5000);
            }
            
            // 保持运行
            try {
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            
            client.disconnect();
            heartbeat.stop();
        }
    }
}

public class DeviceManager {
    
    private Map<String, DeviceInfo> devices;
    private IoTTCPServer server;
    
    public DeviceManager(IoTTCPServer server) {
        this.devices = new ConcurrentHashMap<>();
        this.server = server;
    }
    
    public void registerDevice(String deviceId, Socket socket) {
        DeviceInfo info = new DeviceInfo();
        info.setDeviceId(deviceId);
        info.setSocket(socket);
        info.setConnectTime(System.currentTimeMillis());
        info.setStatus(DeviceStatus.ONLINE);
        
        devices.put(deviceId, info);
    }
    
    public void unregisterDevice(String deviceId) {
        DeviceInfo info = devices.remove(deviceId);
        if (info != null) {
            info.setStatus(DeviceStatus.OFFLINE);
        }
    }
    
    public void sendCommand(String deviceId, String command) {
        server.sendToDevice(deviceId, command);
    }
    
    public List<DeviceInfo> getAllDevices() {
        return new ArrayList<>(devices.values());
    }
    
    public DeviceInfo getDevice(String deviceId) {
        return devices.get(deviceId);
    }
}

// 设备信息
public class DeviceInfo {
    private String deviceId;
    private Socket socket;
    private DeviceStatus status;
    private long connectTime;
    private long lastHeartbeatTime;
    
    // Getters and setters
    public String getDeviceId() { return deviceId; }
    public void setDeviceId(String deviceId) { this.deviceId = deviceId; }
    public Socket getSocket() { return socket; }
    public void setSocket(Socket socket) { this.socket = socket; }
    public DeviceStatus getStatus() { return status; }
    public void setStatus(DeviceStatus status) { this.status = status; }
    public long getConnectTime() { return connectTime; }
    public void setConnectTime(long connectTime) { this.connectTime = connectTime; }
    public long getLastHeartbeatTime() { return lastHeartbeatTime; }
    public void setLastHeartbeatTime(long lastHeartbeatTime) { this.lastHeartbeatTime = lastHeartbeatTime; }
}

enum DeviceStatus {
    ONLINE,
    OFFLINE,
    ERROR
}

import java.util.concurrent.*;

public class ConnectionPool {
    
    private BlockingQueue<Socket> pool;
    private int maxSize;
    private String host;
    private int port;
    
    public ConnectionPool(String host, int port, int maxSize) {
        this.host = host;
        this.port = port;
        this.maxSize = maxSize;
        this.pool = new LinkedBlockingQueue<>(maxSize);
        
        // 初始化连接池
        initializePool();
    }
    
    private void initializePool() {
        for (int i = 0; i < maxSize; i++) {
            try {
                Socket socket = new Socket(host, port);
                pool.offer(socket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public Socket borrowConnection() throws InterruptedException {
        Socket socket = pool.take();
        if (socket.isClosed() || !socket.isConnected()) {
            // 重新创建连接
            try {
                socket = new Socket(host, port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return socket;
    }
    
    public void returnConnection(Socket socket) {
        if (socket != null && !socket.isClosed()) {
            pool.offer(socket);
        }
    }
}

import java.util.concurrent.*;

public class AsyncMessageProcessor {
    
    private ExecutorService executor;
    private BlockingQueue<IoTMessage> messageQueue;
    
    public AsyncMessageProcessor(int threadCount) {
        this.executor = Executors.newFixedThreadPool(threadCount);
        this.messageQueue = new LinkedBlockingQueue<>();
        
        // 启动处理线程
        for (int i = 0; i < threadCount; i++) {
            executor.submit(this::processMessages);
        }
    }
    
    public void submit(IoTMessage message) {
        messageQueue.offer(message);
    }
    
    private void processMessages() {
        while (true) {
            try {
                IoTMessage message = messageQueue.take();
                handleMessage(message);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    private void handleMessage(IoTMessage message) {
        // 处理消息
        switch (message.getType()) {
            case DATA:
                processData(message);
                break;
            case COMMAND:
                processCommand(message);
                break;
            case HEARTBEAT:
                processHeartbeat(message);
                break;
        }
    }
    
    private void processData(IoTMessage message) {
        // 处理数据消息
    }
    
    private void processCommand(IoTMessage message) {
        // 处理命令消息
    }
    
    private void processHeartbeat(IoTMessage message) {
        // 处理心跳消息
    }
}