Açıklama Yok
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

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415
  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. if(context.ssrcs.has(key)){
  65. //console.log(context.ssrcs.get(key));
  66. context.ssrcs.delete(key);
  67. Logger.log("ssrc session removed ",ssrc);
  68. }
  69. });
  70. //ntv add 上一个函数主动关闭,或网络异常,或其他原因关闭rtmp
  71. context.nodeEvent.on('rtmpClientClose', (ssrc) => {
  72. var key = '' + ssrc;
  73. if (context.publishers.has(key)) {
  74. context.publishers.delete(key);
  75. Logger.log("rtmp session removed ",ssrc);
  76. }
  77. });
  78. }
  79. //接收到 INVITE SDP 描述
  80. sdpReceived(sdp) {
  81. //判断流发送者SDP描述,如果是 TCP主动模式 则创建主动取流客户端
  82. if (sdp.media.length > 0 && sdp.media[0].ssrc) {
  83. let ssrc = sdp.media[0].ssrc;
  84. let host = sdp.connection.ip || sdp.origin.address;
  85. let version = sdp.connection.version || 4; //IPV4 or IPV6
  86. let port = sdp.media[0].port;
  87. let protocol = sdp.media[0].protocol;
  88. let mode = 0;
  89. switch (protocol) {
  90. //UDP
  91. case 'RTP/AVP':
  92. {
  93. mode = 0;
  94. }
  95. break;
  96. //TCP
  97. case 'TCP/RTP/AVP':
  98. {
  99. let setup = sdp.media[0].setup;
  100. switch (setup) {
  101. //背动模式,需要创建TCP-Client 去取流
  102. case 'passive':
  103. {
  104. mode = 2;
  105. this.createTCPClient(ssrc, host, port);
  106. }
  107. break;
  108. //主动模式
  109. case 'active':
  110. {
  111. mode = 1;
  112. }
  113. break;
  114. }
  115. }
  116. break;
  117. }
  118. }
  119. }
  120. //创建TCP主动取流客户端
  121. //ntv 该逻辑没有被使用
  122. createTCPClient(ssrc, host, port) {
  123. if (!this.tcpClients[ssrc]) {
  124. let tcpClient = new Net.Socket();
  125. this.tcpClients[ssrc] = tcpClient;
  126. tcpClient._cache = Buffer.alloc(0);
  127. tcpClient.connect(port, host, () => { Logger.log("[GB28181_TCP_Active] 连接成功,等待接收 RTP 数据包...") });
  128. tcpClient.on('data', (data) => {
  129. tcpClient._cache = Buffer.concat([tcpClient._cache, data]);
  130. while (tcpClient._cache.length > 1 && tcpClient._cache.length >= (tcpClient._cache.readUInt16BE(0) + 2)) {
  131. let rtplength = tcpClient._cache.readUInt16BE(0);
  132. let rtpData = tcpClient._cache.slice(2, rtplength + 2);
  133. NodeRtpSession.parseRTPacket(rtpData);
  134. tcpClient._cache = tcpClient._cache.slice(rtplength + 2);
  135. }
  136. });
  137. //连接关闭
  138. tcpClient.on('error', (err) => {
  139. Logger.log("[GB28181_TCP_Active] 连接关闭...") ;
  140. tcpClient.destroy();
  141. delete this.tcpClients[ssrc];
  142. });
  143. }
  144. }
  145. //TCPServer/UDPServer 接收到nalus
  146. rtpReceived(ssrc, timestamp, packet) {
  147. if (!context.publishers.has(ssrc)) {
  148. var rtmpClient = new NodeRtmpClient(`${this.rtmpServer}/${ssrc}`);
  149. rtmpClient.startPush();
  150. Logger.log("Publish to rtmp server ",this.rtmpServer,ssrc);
  151. //RTMP 发布流状态
  152. rtmpClient.on('status', (info) => {
  153. if (info.code === 'NetStream.Publish.Start'){
  154. rtmpClient.isPublishStart = true;
  155. Logger.log("rtmp stream start ",ssrc);
  156. }
  157. });
  158. //连接关闭
  159. rtmpClient.on('close', () => {
  160. Logger.log("rtmp stream closed ",ssrc);
  161. context.nodeEvent.emit('rtmpClientClose', ssrc);
  162. });
  163. //ntv 同时使用string key
  164. var key = '' + ssrc;
  165. context.publishers.set(key, rtmpClient);
  166. }
  167. let rtmpClinet = context.publishers.get(ssrc);
  168. //记录收包时间,长时间未收包关闭会话
  169. rtmpClinet._lastReceiveTime = new Date();
  170. //发送视频第一包
  171. if (!rtmpClinet.sendfirstVideoPacket && rtmpClinet.isPublishStart) {
  172. let streaminfo = rtmpClinet._streaminfo;
  173. switch (streaminfo.video) {
  174. case 0x24:
  175. {
  176. let vps = rtmpClinet._vps;
  177. let sps = rtmpClinet._sps;
  178. let pps = rtmpClinet._pps;
  179. if (vps && sps && pps) {
  180. 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]);
  181. rtmpClinet.pushVideo(_packet, 0);
  182. rtmpClinet.deltaVideo = 0;
  183. rtmpClinet.sendfirstVideoPacket = true;
  184. }
  185. }
  186. break;
  187. case 0x1b:
  188. {
  189. let sps = rtmpClinet._sps;
  190. let pps = rtmpClinet._pps;
  191. if (sps && pps) {
  192. 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]);
  193. rtmpClinet.pushVideo(_packet, 0);
  194. rtmpClinet.deltaVideo = 0;
  195. rtmpClinet.sendfirstVideoPacket = true;
  196. }
  197. }
  198. break;
  199. }
  200. }
  201. //发送音频第一包
  202. if (!rtmpClinet.sendfirstAudioPacket && rtmpClinet.isPublishStart) {
  203. let streaminfo = rtmpClinet._streaminfo;
  204. switch (streaminfo.audio) {
  205. //AAC
  206. case 0x0f:
  207. {
  208. //ToDo 需要判断音频信息,采样率,采样深度,码率
  209. }
  210. break;
  211. //G711a
  212. case 0x90:
  213. {
  214. var _packet = Buffer.from([0x70]);
  215. rtmpClinet.pushAudio(_packet, 0);
  216. rtmpClinet.deltaAudio = 0;
  217. rtmpClinet.sendfirstAudioPacket = true;
  218. }
  219. break;
  220. //G711u
  221. case 0x91:
  222. {
  223. var _packet = Buffer.from([0x80]);
  224. rtmpClinet.pushAudio(_packet, 0);
  225. rtmpClinet.deltaAudio = 0;
  226. rtmpClinet.sendfirstAudioPacket = true;
  227. }
  228. break;
  229. }
  230. }
  231. //判断packet.streaminfo H264/H265
  232. if (!rtmpClinet._streaminfo && packet.streaminfo)
  233. rtmpClinet._streaminfo = packet.streaminfo;
  234. if (!rtmpClinet._streaminfo.video && packet.streaminfo.video)
  235. rtmpClinet._streaminfo.video = packet.streaminfo.video;
  236. if (!rtmpClinet._streaminfo.audio && packet.streaminfo.audio)
  237. rtmpClinet._streaminfo.audio = packet.streaminfo.audio;
  238. //发送视频
  239. packet.video.forEach(nalu => {
  240. switch (rtmpClinet._streaminfo.video) {
  241. //H265
  242. case 0x24:
  243. {
  244. let naluType = (nalu.readUInt8(0) & 0x7E) >> 1;
  245. switch (naluType) {
  246. case 19:
  247. rtmpClinet._keyframe = nalu;
  248. break;
  249. case 32:
  250. if (!rtmpClinet._vps)
  251. rtmpClinet._vps = nalu;
  252. break;
  253. case 33:
  254. if (!rtmpClinet._sps)
  255. rtmpClinet._sps = nalu;
  256. break;
  257. case 34:
  258. if (!rtmpClinet._pps)
  259. rtmpClinet._pps = nalu;
  260. break;
  261. }
  262. //flv封装
  263. if (naluType !== 32 && naluType !== 33 && naluType !== 34) {
  264. 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]);
  265. rtmpClinet.deltaVideo += timestamp / 90;
  266. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstVideoPacket)
  267. rtmpClinet.pushVideo(packet, rtmpClinet.deltaVideo);
  268. }
  269. }
  270. break;
  271. //H264
  272. case 0x1b:
  273. {
  274. let naluType = nalu.readUInt8(0) & 0x1F;
  275. switch (naluType) {
  276. case 5:
  277. rtmpClinet._keyframe = nalu;
  278. break;
  279. case 7:
  280. if (!rtmpClinet._sps)
  281. rtmpClinet._sps = nalu;
  282. break;
  283. case 8:
  284. if (!rtmpClinet._pps)
  285. rtmpClinet._pps = nalu;
  286. break;
  287. }
  288. //flv封装
  289. if (naluType !== 7 && naluType !== 8) {
  290. 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]);
  291. rtmpClinet.deltaVideo += timestamp / 90;
  292. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstVideoPacket)
  293. rtmpClinet.pushVideo(packet, rtmpClinet.deltaVideo);
  294. }
  295. }
  296. break;
  297. //SVAC
  298. case 0x80:
  299. break;
  300. }
  301. });
  302. //发送音频
  303. if (packet.audio.length > 0) {
  304. if (rtmpClinet.isPublishStart && rtmpClinet.sendfirstAudioPacket) {
  305. switch (rtmpClinet._streaminfo.audio) {
  306. //G711a
  307. case 0x90:
  308. {
  309. rtmpClinet.deltaAudio += (packet.audio.length / 8000) * 1000;
  310. rtmpClinet.pushAudio(Buffer.concat([Buffer.from([0x70]), packet.audio]), rtmpClinet.deltaAudio);
  311. }
  312. break;
  313. //G711u
  314. case 0x91:
  315. {
  316. rtmpClinet.deltaAudio += (packet.audio.length / 8000) * 1000;
  317. rtmpClinet.pushAudio(Buffer.concat([Buffer.from([0x80]), packet.audio]), rtmpClinet.deltaAudio);
  318. }
  319. break;
  320. }
  321. }
  322. }
  323. }
  324. stop() {
  325. if (this.tcpServer)
  326. this.tcpServer.close();
  327. if (this.udpServer)
  328. this.udpServer.close();
  329. context.sessions.forEach((session, id) => {
  330. if (session instanceof NodeRtpSession) {
  331. session.socket.destroy();
  332. context.sessions.delete(id);
  333. }
  334. });
  335. }
  336. }
  337. module.exports = NodeGB28181StreamServer