ntrip-server/app/middleware/ntripCaster/ntripCaster.js

const net = require('net');
const CryptoJS = require("crypto-js");

const KEY = CryptoJS.enc.Utf8.parse('BDQH');

class SimpleNtripCaster {
    constructor(app) {
        this.app = app;
        const { config, redis } = app;
        const port = config.caster.port;
        this.port = port;
        this.redis = redis;
        this.sourcetable = [];
        this.serverSockets = new Map(); // mountpoint -> server socket
        this.sizeSockets = new Map(); // client socket -> mountpoint
        this.clientSockets = new Map(); // client socket -> mountpoint
        this.mountData = new Map(); // mountpoint -> 当前连接的 server socket
        console.log(`[Caster] 启动在端口 ${port}`);
    }

    start() {
        this.server = net.createServer((socket) => {
            let clientInfo = '';
            socket.on('data', (data) => {
                const request = data.toString();
                clientInfo += request;
                // 检查是否收到完整的HTTP请求（以空行结束）
                if (clientInfo.includes('\r\n\r\n')) {
                    this.handleRequest(socket, clientInfo);
                    clientInfo = '';
                }
            });
            socket.on('end', () => {
                console.log('[Caster] 连接关闭');
                this.removeClient(socket);
                this.broadSizeData();
            });
            socket.on('error', (err) => {
                console.log('[Caster] 连接错误:', err.message);
                this.removeClient(socket);
                this.broadSizeData();
            });
        });
        this.server.listen(this.port, () => {
            console.log(`[Caster] 等待连接...`);
        });
    }

    sendSourcetable(socket, statusCode, statusMessage, headers1 = {}, headers2 = {}, body = '') {
        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
        for (const [key, value] of Object.entries(headers1)) {
            response += `${key}: ${value}\r\n`;
        }
        response += `\r\n`;
        response += `SOURCETABLE ${statusCode} ${statusMessage}\r\n`;
        for (const [key, value] of Object.entries(headers2)) {
            response += `${key}: ${value}\r\n`;
        }
        response += '\r\n' + body;
        socket.end(response);
    }

    sendResponse(socket, statusCode, statusMessage, headers = {}, body = '') {
        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
        for (const [key, value] of Object.entries(headers)) {
            response += `${key}: ${value}\r\n`;
        }
        response += '\r\n' + body;
        socket.end(response);
    }

    clientSendResponse(socket, statusCode, statusMessage, headers = {}, body = '') {
        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
        for (const [key, value] of Object.entries(headers)) {
            response += `${key}: ${value}\r\n`;
        }
        response += '\r\n' + body;
        socket.write(response);
    }

    handleOptions(socket, headers) {
        this.sendResponse(socket, 204, 'No Content', {
            'Access-Control-Allow-Origin': headers.origin,
            'Access-Control-Allow-Methods': 'GET, OPTIONS, SIZE, SOURCE',
            'Access-Control-Allow-Headers': 'Content-Type, uid, Authorization, ntrip-version, Client-Type, Request-Type',
        });
    }

    parseHeaders(headerLines) {
        const headers = {};
        headerLines.forEach((line) => {
            const [key, value] = line.split(': ');
            headers[key.toLowerCase()] = value;
        });
        return headers;
    }

    handleRequest(socket, request) {
        const lines = request.split('\r\n');
        const firstLine = lines[0];
        console.log(`[Caster] 收到请求: ${firstLine.substring(0, 50)}...`);
        // 处理 OPTIONS 请求
        if (firstLine.split(' ')[0] === 'OPTIONS') {
            const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
            this.handleOptions(socket, headers);
        } else if (firstLine === 'GET / HTTP/1.1' || firstLine === 'GET / HTTP/1.0' || firstLine === 'GET /') {
            // 客户端请求源列表
            this.handleSourcetable(socket, lines);
        } else if (firstLine.startsWith('SOURCE')) {
            // Server 挂载请求
            this.handleServerMount(socket, lines);
        } else if (firstLine.startsWith('GET /')) {
            // 客户端请求数据流
            this.handleClientRequest(socket, lines);
        } else {
            this.sendResponse(socket, 400, 'Bad Request', {
                'Access-Control-Allow-Origin': '*'
            });
        }
    }

    // 发送源列表
    handleSourcetable(socket, lines) {
        const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
        let table = '';
        this.sourcetable.forEach(element => {
            // SourceID 数据源唯一标识
            // StreamFormat 0=RTCM 2, 1=RTCM 3.0, 2=RTCM 3.1/3.2, ... 
            // Details 位掩码，支持的 RTCM 消息类型
            // StreamType 1=OBS, 2=NAV, 3=STA, 4=COR
            // Carrier 载波相位类型 
            // NavSys 导航系统位掩码 
            // Net 网络标识符 
            // Country ISO 3166 数字代码
            // Latitude 纬度
            // Longitude 经度
            // NMEA NMEA 消息支持
            // Solution 解算类型
            // Compressed Y/N 是否压缩
            // Framed Y/N 是否帧格式
            // MSM MSM 支持掩码
            const { mountpoint, SourceAgent, MountpointOptions } = element;
            let SourceID, Format, FormatDetails, StreamType, Carrier, NavSystem, NetWork, Country, Latitude, Longitude, NMEA, Soution, Compressed, Framed, MSM, Description, StreamID, Generator, Compression, Authenticated, Fee, Bitrate;
            if (SourceAgent == 'NTRIP RTKLIB/demo5_b34L_pre1') {
                // 源ID
                SourceID = MountpointOptions[0] == '' ? '' : MountpointOptions[0];
                // 格式
                Format = MountpointOptions[1] == '' ? 'RTCM 3.2' : MountpointOptions[1];
                // 格式详情
                FormatDetails = MountpointOptions[2] == '' ? '1005(1),1033(1),1074(1),1084(1),1094(1),1114(1),1124(1)' : MountpointOptions[2];
                // 载体
                Carrier = MountpointOptions[3] == '' ? 2 : MountpointOptions[3];
                // 导航系统
                NavSystem = MountpointOptions[4] == '' ? 'GPS+GLO+GAL+BDS+QZS' : MountpointOptions[4];
                // 网络
                NetWork = MountpointOptions[5] == '' ? '' : MountpointOptions[5];
                // 国家
                Country = MountpointOptions[6] == '' ? '' : MountpointOptions[6];
                // 纬度
                Latitude = MountpointOptions[7] == '' ? 0.0 : MountpointOptions[7];
                // 经度
                Longitude = MountpointOptions[8] == '' ? 0.0 : MountpointOptions[8];
                // NMEA 消息支持
                NMEA = MountpointOptions[9] == '' ? 0 : MountpointOptions[9];
                // 解算类型
                Soution = MountpointOptions[10] == '' ? 0 : MountpointOptions[10];
                // 发生器
                Generator = MountpointOptions[11] == '' ? 'B' : MountpointOptions[11];
                // 压缩加密
                Compression = MountpointOptions[12] == '' ? 'N' : MountpointOptions[12];
                // 认证
                Authenticated = MountpointOptions[13] == '' ? 'N' : MountpointOptions[13];
                // 费用
                Fee = MountpointOptions[14] == '' ? 'N' : MountpointOptions[14];
                // 比特率
                Bitrate = MountpointOptions[15] == '' ? '115200' : MountpointOptions[15];
                // 描述
                Description = 'RTK';
                // 流ID
                StreamID = 'none';
            } else {
                // 源ID
                SourceID = MountpointOptions[0] == '' ? '' : MountpointOptions[0];
                // 格式
                Format = MountpointOptions[1] == '' ? 'RTCM 3.2' : MountpointOptions[1];
                // 格式详情
                FormatDetails = MountpointOptions[2] == '' ? '1005(1),1033(1),1074(1),1084(1),1094(1),1114(1),1124(1)' : MountpointOptions[2];
                StreamType = MountpointOptions[3] == '' ? '' : MountpointOptions[3];
                // 载体
                Carrier = MountpointOptions[4] == '' ? 2 : MountpointOptions[4];
                // 导航系统
                NavSystem = MountpointOptions[5] == '' ? 'GPS+GLO+GAL+BDS+QZS' : MountpointOptions[5];
                // 网络
                NetWork = MountpointOptions[6] == '' ? '' : MountpointOptions[6];
                // 国家
                Country = MountpointOptions[7] == '' ? '' : MountpointOptions[7];
                // 纬度
                Latitude = MountpointOptions[8] == '' ? 0.0 : MountpointOptions[8];
                // 经度
                Longitude = MountpointOptions[9] == '' ? 0.0 : MountpointOptions[9];
                // NMEA
                NMEA = MountpointOptions[10] == '' ? 0 : MountpointOptions[10];
                // 解决方案
                Soution = MountpointOptions[11] == '' ? 0 : MountpointOptions[11];
                // 描述
                Description = 'RTK';
                // 流ID
                StreamID = 'none';
                // 发生器
                Generator = 'B';
                // 压缩加密
                Compression = 'N';
            }
            const found = lines.find(line => line.startsWith('Ntrip-Version: Ntrip/2.0'));
            if (found == 'Ntrip-Version: Ntrip/2.0') {
                if (SourceAgent == 'NTRIP RTKLIB/demo5_b34L_pre1') {
                    table += `STR;${mountpoint};${SourceID};${Format};${FormatDetails};${Carrier};${NavSystem};${NetWork};${Country};${Latitude};${Longitude};${NMEA};${Soution};${Description};${StreamID};${Generator};${Compression};${Authenticated};${Fee};${Bitrate};none\r\n`;
                } else {
                    // 变体B：RTCM/ntrip-2风格
                    // table += `STR;${Format};${mountpoint};${FormatDetails};;${Carrier};${NavSystem};${NetWork};${Country};${Latitude};${Longitude};${NMEA};${SourceID};;;;\r\n`;
                    table += `STR;${Format};${SourceID};${mountpoint};${Format};${FormatDetails};${Carrier};${NavSystem};${NetWork};${Country};${Latitude};${Longitude};${NMEA};${Soution};${Description};${StreamID};${Generator};${Compression};${Authenticated};${Fee};${Bitrate};0\r\n`;
                }
            } else {
                // 变体A：BKG/NTRIP 1.0风格
                // table += `STR;${Format};${mountpoint};${FormatDetails};;${Carrier};${NavSystem};${SourceID};${Latitude};${Longitude};0.0;${NetWork};${Country};;;;\r\n`;
                table += `STR;${Format};${mountpoint};${Format};${FormatDetails};${Carrier};${NavSystem};${NetWork};${Country};${Latitude};${Longitude};${NMEA};${Soution};${Description};${StreamID};${Generator};${Compression};0\r\n`;
            }
        });
        table += 'ENDSOURCETABLE\r\n';
        this.sendSourcetable(socket, 200, 'OK', {
            'Access-Control-Allow-Origin': headers.origin ? headers.origin : '*',
            'Server': 'NTRIP Caster',
            'Content-Type': 'text/plain',
            'Date': 'Tue, 12 May 2026 08:00:00 GMT',
            'Connection': 'close'
        }, {
            'Access-Control-Allow-Origin': headers.origin ? headers.origin : '*',
            'Server': 'Caster/1.0',
            'Content-Type': 'text/plain',
            'Charset': 'UTF-8'
        }, table);
        console.log('[Caster] 已发送源列表');
    }

    // 处理 Server 挂载
    async handleServerMount(socket, lines) {
        const source = lines[0].split(' ');
        // 密码
        const password = source[1];
        // 挂载点
        const mountpoint = source[2];
        const findMountpoint = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/findMountpoint', {
            dataType: 'json',
            contentType: 'application/x-www-form-urlencoded',
            method: 'POST',
            timeout: 3000,
            data: {
                mountpoint: mountpoint
            }
        });
        if (findMountpoint.data.code == 404) {
            this.sendResponse(socket, 404, 'Mountpoint Not Found', {});
            return;
        }
        if (password !== findMountpoint.data.password) {
            this.sendResponse(socket, 401, 'Unauthorized', {});
            return;
        }
        const SourceAgent = lines[1].split(': ')[1];
        const MountpointOptions = lines[2].split(': ')[1].split(';');
        this.sourcetable.push({ password, mountpoint, SourceAgent, MountpointOptions });
        this.serverSockets.set(mountpoint, socket);
        this.mountData.set(mountpoint, {
            socket: socket,
            lastUpdate: Date.now()
        });
        // 响应 Server
        socket.write('ICY 200 OK\r\n\r\n');
        console.log(`[Caster] Server 挂载成功: ${mountpoint}`);
        // 监听 Server 数据
        socket.on('data', (data) => {
            this.broadcastData(mountpoint, data);
        });
        socket.on('end', () => {
            console.log(`[Caster] Server 断开: ${mountpoint}`);
            this.serverSockets.delete(mountpoint);
            this.mountData.delete(mountpoint);
        });
    }

    // 处理 Client 请求
    async handleClientRequest(socket, lines) {
        // 解析请求的挂载点
        const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
        const firstLine = lines[0];
        const match = firstLine.match(/GET \/(\w+) HTTP\/1\.[01]/);
        if (!match) {
            this.sendResponse(socket, 404, '404 Not Found', {
                'Access-Control-Allow-Origin': headers.origin,
            });
            return;
        }
        const mountpoint = match[1];
        console.log(`[Caster] Client 请求挂载点: ${mountpoint}`);
        const find = lines.find(line => line.startsWith('Request-Type: SIZE'));
        const found = lines.find(line => line.startsWith('Client-Type: Virtual-NTRIP-Client/1.0'));
        if (found !== 'Client-Type: Virtual-NTRIP-Client/1.0') {
            if (find !== 'Request-Type: SIZE') {
                const findMountpoint = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/findMountpoint', {
                    dataType: 'json',
                    contentType: 'application/x-www-form-urlencoded',
                    method: 'POST',
                    timeout: 3000,
                    data: {
                        mountpoint: mountpoint
                    }
                });
                if (findMountpoint.data.code == 404) {
                    this.sendResponse(socket, 404, 'Mountpoint Not Found', {});
                    return;
                }
                const getPassword = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/getPassword', {
                    dataType: 'json',
                    contentType: 'application/x-www-form-urlencoded',
                    method: 'POST',
                    timeout: 3000,
                    data: {
                        mountpoint: mountpoint,
                        base: headers.authorization.split(' ')[1],
                        status: findMountpoint.data.status
                    }
                });
                if (getPassword.data.code == 404) {
                    this.sendResponse(socket, 401, 'Unauthorized', {});
                    return;
                }
            }
        }
        if (find !== 'Request-Type: SIZE') {
            if (!this.mountData.has(mountpoint)) {
                this.sendResponse(socket, 404, 'Mountpoint Not Found', {
                    'Access-Control-Allow-Origin': '*'
                });
                return;
            }
        }
        if (found == 'Client-Type: Virtual-NTRIP-Client/1.0') {
            // 开发/测试环境
            const uid = lines[7].split(': ')[1];
            const token = lines[6].split(': ')[1];
            // 生产环境
            // const uid = lines[9].split(': ')[1];
            // const token = lines[5].split(': ')[1];
            let decryptUid = CryptoJS.AES.decrypt(uid, KEY, {
                mode: CryptoJS.mode.ECB,
                padding: CryptoJS.pad.Pkcs7
            }).toString(CryptoJS.enc.Utf8).split('?')[0];
            const redisType = {
                "B": "db0",
                "C": "db1",
                "T": "db5",
            };
            const lastCode = decryptUid[decryptUid.length - 1];
            const rdmcode = await this.redis.get(redisType[lastCode]).get(token);
            if (!rdmcode) {
                this.sendResponse(socket, 401, 'Unauthorized', {
                    'Access-Control-Allow-Origin': headers.origin
                });
                return;
            }
        }
        if (find == 'Request-Type: SIZE') {
            const uid = lines[8].split(': ')[1];
            // 接受客户端连接
            this.sizeSockets.set(uid + ':' + mountpoint, socket);
            const size = [...this.clientSockets.values()].filter(value => value == mountpoint);
            this.clientSendResponse(socket, 200, 'OK', {
                'Access-Control-Allow-Origin': headers.origin, // 设置允许的源
                'Content-Type': 'text/event-stream',
                'Connection': 'keep-alive'
            }, size.length);
            console.log(`[Caster] 已发送Client挂载数量`);
        } else {
            // 接受客户端连接
            this.clientSockets.set(socket, mountpoint);
            this.clientSendResponse(socket, 200, 'OK', {
                'Access-Control-Allow-Origin': headers.origin, // 设置允许的源
                'Content-Type': 'text/event-stream',
                'Connection': 'keep-alive'
            });
            this.broadSizeData();
            console.log(`[Caster] Client 已连接至: ${mountpoint}`);
        }
    }

    // 广播已连接到此挂载点的客户端数量数据给所有已连接到此挂载点的客户端
    broadSizeData() {
        for (const [clientMsg, client] of this.sizeSockets.entries()) {
            const mountpoint = clientMsg.split(':')[1];
            const size = [...this.clientSockets.values()].filter(value => value == mountpoint);
            client.write(size.length + '');
        }
    }

    // 广播数据给所有连接到此挂载点的客户端
    broadcastData(mountpoint, data) {
        for (const [client, clientMount] of this.clientSockets.entries()) {
            if (clientMount === mountpoint && !client.destroyed) {
                client.write(data);
            }
        }
    }

    removeClient(socket) {
        this.clientSockets.delete(socket);
    }

    findHeader(lines, headerName) {
        for (const line of lines) {
            if (line.startsWith(headerName + ':')) {
                return line.substring(headerName.length + 1).trim();
            }
        }
        return null;
    }
}

// 导出类供单独使用
module.exports = { SimpleNtripCaster };