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 14KB

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