425 lines
19 KiB
JavaScript
425 lines
19 KiB
JavaScript
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];
|
||
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 }; |