Geen omschrijving
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

server.js 15KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410
  1. const Net = require('net');
  2. const Logger = require('../core/logger');
  3. const NodeRtpSession = require('./session');
  4. const context = require('../core/ctx');
  5. const NodeRtmpClient = require('../core/rtmp_client');
  6. const RtpSession = require("rtp-rtcp").RtpSession;
  7. //GB28181 媒体服务器
  8. class NodeGB28181StreamServer {
  9. constructor(config) {
  10. this.listen = config.GB28181.streamServer.listen || 9200;
  11. this.host = config.GB28181.streamServer.host || '0.0.0.0';
  12. //开启端口复用
  13. if (config.GB28181.streamServer.invite_port_fixed) {
  14. //RTP-RTCP UDP-Server
  15. this.udpServer = new RtpSession(this.listen);
  16. this.udpServer.createRtcpServer();
  17. //TCP-Server
  18. this.tcpServer = Net.createServer((socket) => {
  19. let session = new NodeRtpSession(config, socket);
  20. session.run();
  21. });
  22. }
  23. //主动取流客户端(TCP/主动模式)
  24. this.tcpClients = {};
  25. //默认的RTMP服务器基地址
  26. this.rtmpServer = config.GB28181.streamServer.rtmpServer || 'rtmp://127.0.0.1/live';
  27. }
  28. run() {
  29. if (this.udpServer) {
  30. //TCP
  31. this.tcpServer.listen(this.listen, () => {
  32. Logger.log(`GB28181-Media Server started on TCP port: ${this.listen}`);
  33. });
  34. this.tcpServer.on('error', (e) => {
  35. Logger.error(`GB28181-Media Server ${e}`);
  36. });
  37. this.tcpServer.on('close', () => {
  38. Logger.log('GB28181-Media Server Close.');
  39. });
  40. }
  41. if (this.udpServer) {
  42. //UDP
  43. this.udpServer.on("listening", () => {
  44. Logger.log(`GB28181-Media Server started on UDP port: ${this.listen}`);
  45. });
  46. this.udpServer.on("message", (msg, info) => {
  47. NodeRtpSession.parseRTPacket(msg);
  48. });
  49. }
  50. //SDP收到
  51. context.nodeEvent.on('sdpReceived', this.sdpReceived.bind(this));
  52. //RTP己处理好
  53. context.nodeEvent.on('rtpReadyed', this.rtpReceived.bind(this));
  54. //停止播放,关闭推流客户端(rtp客户端主动关闭连接)
  55. context.nodeEvent.on('stopPlayed', (ssrc) => {
  56. //ntv mod 转换成string,否者检索不到,ssrc传入的时number
  57. var key = '' + ssrc;
  58. if (context.publishers.has(key)) {
  59. let rtmpClient = context.publishers.get(key);
  60. rtmpClient.stop();
  61. context.publishers.delete(key);
  62. Logger.log("rtmp session removed ",ssrc);
  63. }
  64. });
  65. //ntv add 上一个函数主动关闭,或网络异常,或其他原因关闭rtmp
  66. context.nodeEvent.on('rtmpClientClose', (ssrc) => {
  67. var key = '' + ssrc;
  68. if (context.publishers.has(key)) {
  69. context.publishers.delete(key);
  70. Logger.log("rtmp session removed ",ssrc);
  71. }
  72. });
  73. }
  74. //接收到 INVITE SDP 描述
  75. sdpReceived(sdp) {
  76. //判断流发送者SDP描述,如果是 TCP主动模式 则创建主动取流客户端
  77. if (sdp.media.length > 0 && sdp.media[0].ssrc) {
  78. let ssrc = sdp.media[0].ssrc;
  79. let host = sdp.connection.ip || sdp.origin.address;
  80. let version = sdp.connection.version || 4; //IPV4 or IPV6
  81. let port = sdp.media[0].port;
  82. let protocol = sdp.media[0].protocol;
  83. let mode = 0;
  84. switch (protocol) {
  85. //UDP
  86. case 'RTP/AVP':
  87. {
  88. mode = 0;
  89. }
  90. break;
  91. //TCP
  92. case 'TCP/RTP/AVP':
  93. {
  94. let setup = sdp.media[0].setup;
  95. switch (setup) {
  96. //背动模式,需要创建TCP-Client 去取流
  97. case 'passive':
  98. {
  99. mode = 2;
  100. this.createTCPClient(ssrc, host, port);
  101. }
  102. break;
  103. //主动模式
  104. case 'active':
  105. {
  106. mode = 1;
  107. }
  108. break;
  109. }
  110. }
  111. break;
  112. }
  113. }
  114. }
  115. //创建TCP主动取流客户端
  116. //ntv 该逻辑没有被使用
  117. createTCPClient(ssrc, host, port) {
  118. if (!this.tcpClients[ssrc]) {
  119. let tcpClient = new Net.Socket();
  120. this.tcpClients[ssrc] = tcpClient;
  121. tcpClient._cache = Buffer.alloc(0);
  122. tcpClient.connect(port, host, () => { Logger.log("[GB28181_TCP_Active] 连接成功,等待接收 RTP 数据包...") });
  123. tcpClient.on('data', (data) => {
  124. tcpClient._cache = Buffer.concat([tcpClient._cache, data]);
  125. while (tcpClient._cache.length > 1 && tcpClient._cache.length >= (tcpClient._cache.readUInt16BE(0) + 2)) {
  126. let rtplength = tcpClient._cache.readUInt16BE(0);
  127. let rtpData = tcpClient._cache.slice(2, rtplength + 2);
  128. NodeRtpSession.parseRTPacket(rtpData);
  129. tcpClient._cache = tcpClient._cache.slice(rtplength + 2);
  130. }
  131. });
  132. //连接关闭
  133. tcpClient.on('error', (err) => {
  134. Logger.log("[GB28181_TCP_Active] 连接关闭...") ;
  135. tcpClient.destroy();
  136. delete this.tcpClients[ssrc];
  137. });
  138. }
  139. }
  140. //TCPServer/UDPServer 接收到nalus
  141. rtpReceived(ssrc, timestamp, packet) {
  142. if (!context.publishers.has(ssrc)) {
  143. var rtmpClient = new NodeRtmpClient(`${this.rtmpServer}/${ssrc}`);
  144. rtmpClient.startPush();
  145. Logger.log("Publish to rtmp server ",this.rtmpServer,ssrc);
  146. //RTMP 发布流状态
  147. rtmpClient.on('status', (info) => {
  148. if (info.code === 'NetStream.Publish.Start'){
  149. rtmpClient.isPublishStart = true;
  150. Logger.log("rtmp stream start ",ssrc);
  151. }
  152. });
  153. //连接关闭
  154. rtmpClient.on('close', () => {
  155. Logger.log("rtmp stream closed ",ssrc);
  156. context.nodeEvent.emit('rtmpClientClose', ssrc);
  157. });
  158. //ntv 同时使用string key
  159. var key = '' + ssrc;
  160. context.publishers.set(key, rtmpClient);
  161. }
  162. let rtmpClinet = context.publishers.get(ssrc);
  163. //记录收包时间,长时间未收包关闭会话
  164. rtmpClinet._lastReceiveTime = new Date();
  165. //发送视频第一包
  166. if (!rtmpClinet.sendfirstVideoPacket && rtmpClinet.isPublishStart) {
  167. let streaminfo = rtmpClinet._streaminfo;
  168. switch (streaminfo.video) {
  169. case 0x24:
  170. {
  171. let vps = rtmpClinet._vps;
  172. let sps = rtmpClinet._sps;
  173. let pps = rtmpClinet._pps;
  174. if (vps && sps && pps) {
  175. let _packet = Buffer.concat([Buffer.from([0x1C, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x00, 0x00, 0x00, 0x90, 0x00, 0x00, 0x00, 0x00, 0x00, 0xFF, 0x5A, 0xF0, 0x01, 0xFC, 0xFD, 0xF8, 0xF8, 0x00, 0x00, 0x0F, 0x03, 0x20, 0x00, 0x01, vps.length >> 8 & 0xff, vps.length & 0xff]), vps, Buffer.from([0x21, 0x00, 0x01, sps.length >> 8 & 0xff, sps.length & 0xff]), sps, Buffer.from([0x22, 0x00, 0x01, pps.length >> 8 & 0xff, pps.length & 0xff]), pps]);
  176. rtmpClinet.pushVideo(_packet, 0);
  177. rtmpClinet.deltaVideo = 0;
  178. rtmpClinet.sendfirstVideoPacket = true;
  179. }
  180. }
  181. break;
  182. case 0x1b:
  183. {
  184. let sps = rtmpClinet._sps;
  185. let pps = rtmpClinet._pps;
  186. if (sps && pps) {
  187. let _packet = Buffer.concat([Buffer.from([0x17, 0x00, 0x00, 0x00, 0x00, 0x01, sps.readUInt8(1), sps.readUInt8(2), sps.readUInt8(3), 0xff, 0xe1, sps.length >> 8 & 0xff, sps.length & 0xff]), sps, Buffer.from([0x01, pps.length >> 8 & 0xff, pps.length & 0xff]), pps]);
  188. rtmpClinet.pushVideo(_packet, 0);
  189. rtmpClinet.deltaVideo = 0;
  190. rtmpClinet.sendfirstVideoPacket = true;
  191. }
  192. }
  193. break;
  194. }
  195. }
  196. //发送音频第一包
  197. if (!rtmpClinet.sendfirstAudioPacket && rtmpClinet.isPublishStart) {
  198. let streaminfo = rtmpClinet._streaminfo;
  199. switch (streaminfo.audio) {
  200. //AAC
  201. case 0x0f:
  202. {
  203. //ToDo 需要判断音频信息,采样率,采样深度,码率
  204. }
  205. break;
  206. //G711a
  207. case 0x90:
  208. {
  209. var _packet = Buffer.from([0x70]);
  210. rtmpClinet.pushAudio(_packet, 0);
  211. rtmpClinet.deltaAudio = 0;
  212. rtmpClinet.sendfirstAudioPacket = true;
  213. }
  214. break;
  215. //G711u
  216. case 0x91:
  217. {
  218. var _packet = Buffer.from([0x80]);
  219. rtmpClinet.pushAudio(_packet, 0);
  220. rtmpClinet.deltaAudio = 0;
  221. rtmpClinet.sendfirstAudioPacket = true;
  222. }
  223. break;
  224. }
  225. }
  226. //判断packet.streaminfo H264/H265
  227. if (!rtmpClinet._streaminfo && packet.streaminfo)
  228. rtmpClinet._streaminfo = packet.streaminfo;
  229. if (!rtmpClinet._streaminfo.video && packet.streaminfo.video)
  230. rtmpClinet._streaminfo.video = packet.streaminfo.video;
  231. if (!rtmpClinet._streaminfo.audio && packet.streaminfo.audio)
  232. rtmpClinet._streaminfo.audio = packet.streaminfo.audio;
  233. //发送视频
  234. packet.video.forEach(nalu => {
  235. switch (rtmpClinet._streaminfo.video) {
  236. //H265
  237. case 0x24:
  238. {
  239. let naluType = (nalu.readUInt8(0) & 0x7E) >> 1;
  240. switch (naluType) {
  241. case 19:
  242. rtmpClinet._keyframe = nalu;
  243. break;
  244. case 32:
  245. if (!rtmpClinet._vps)
  246. rtmpClinet._vps = nalu;
  247. break;
  248. case 33:
  249. if (!rtmpClinet._sps)
  250. rtmpClinet._sps = nalu;
  251. break;
  252. case 34:
  253. if (!rtmpClinet._pps)
  254. rtmpClinet._pps = nalu;
  255. break;
  256. }
  257. //flv封装
  258. if (naluType !== 32 && naluType !== 33 && naluType !== 34) {
  259. let packet = Buffer.concat([Buffer.from([naluType == 19 ? 0x1C : 0x2C, 0x01, 0x00, 0x00, 0x00, (nalu.length >> 24 & 0xff), (nalu.length >> 16 & 0xff), (nalu.length >> 8 & 0xff), (nalu.length & 0xff)]), nalu]);
  260. rtmpClinet.deltaVideo += timestamp / 90;
  261. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstVideoPacket)
  262. rtmpClinet.pushVideo(packet, rtmpClinet.deltaVideo);
  263. }
  264. }
  265. break;
  266. //H264
  267. case 0x1b:
  268. {
  269. let naluType = nalu.readUInt8(0) & 0x1F;
  270. switch (naluType) {
  271. case 5:
  272. rtmpClinet._keyframe = nalu;
  273. break;
  274. case 7:
  275. if (!rtmpClinet._sps)
  276. rtmpClinet._sps = nalu;
  277. break;
  278. case 8:
  279. if (!rtmpClinet._pps)
  280. rtmpClinet._pps = nalu;
  281. break;
  282. }
  283. //flv封装
  284. if (naluType !== 7 && naluType !== 8) {
  285. let packet = Buffer.concat([Buffer.from([naluType == 5 ? 0x17 : 0x27, 0x01, 0x00, 0x00, 0x00, (nalu.length >> 24 & 0xff), (nalu.length >> 16 & 0xff), (nalu.length >> 8 & 0xff), (nalu.length & 0xff)]), nalu]);
  286. rtmpClinet.deltaVideo += timestamp / 90;
  287. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstVideoPacket)
  288. rtmpClinet.pushVideo(packet, rtmpClinet.deltaVideo);
  289. }
  290. }
  291. break;
  292. //SVAC
  293. case 0x80:
  294. break;
  295. }
  296. });
  297. //发送音频
  298. if (packet.audio.length > 0) {
  299. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstAudioPacket) {
  300. switch (rtmpClinet._streaminfo.audio) {
  301. //G711a
  302. case 0x90:
  303. {
  304. rtmpClinet.deltaAudio += (packet.audio.length / 8000) * 1000;
  305. rtmpClinet.pushAudio(Buffer.concat([Buffer.from([0x70]), packet.audio]), rtmpClinet.deltaAudio);
  306. }
  307. break;
  308. //G711u
  309. case 0x91:
  310. {
  311. rtmpClinet.deltaAudio += (packet.audio.length / 8000) * 1000;
  312. rtmpClinet.pushAudio(Buffer.concat([Buffer.from([0x80]), packet.audio]), rtmpClinet.deltaAudio);
  313. }
  314. break;
  315. }
  316. }
  317. }
  318. }
  319. stop() {
  320. if (this.tcpServer)
  321. this.tcpServer.close();
  322. if (this.udpServer)
  323. this.udpServer.close();
  324. context.sessions.forEach((session, id) => {
  325. if (session instanceof NodeRtpSession) {
  326. session.socket.destroy();
  327. context.sessions.delete(id);
  328. }
  329. });
  330. }
  331. }
  332. module.exports = NodeGB28181StreamServer