From 469717523b4a95b57d0ca214c36ef060b4b12473 Mon Sep 17 00:00:00 2001 From: Li Keqing Date: Fri, 3 Jan 2025 19:32:49 +0800 Subject: [PATCH 1/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BD=93=E6=8B=89?= =?UTF-8?q?=E6=B5=81=E4=BB=A3=E7=90=86=E4=B8=8A=E7=BA=BF=E5=BE=88=E6=85=A2?= =?UTF-8?q?=E6=97=B6=EF=BC=8C=E6=94=B6=E5=88=B0=E6=9D=A5=E8=87=AA=E4=B8=8A?= =?UTF-8?q?=E7=BA=A7=E7=9A=84=20ACK=20=E6=97=A9=E4=BA=8E=E6=B5=81=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=E5=AF=BC=E8=87=B4=E6=8E=A8=E6=B5=81=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../impl/GbChannelPlayServiceImpl.java | 7 +-- .../service/IStreamProxyPlayService.java | 3 +- .../impl/StreamProxyPlayServiceImpl.java | 55 ++++++++++++++++++- 3 files changed, 56 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelPlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelPlayServiceImpl.java index 4d452454..cfa63fd2 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelPlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelPlayServiceImpl.java @@ -129,12 +129,7 @@ public class GbChannelPlayServiceImpl implements IGbChannelPlayService { public void playProxy(CommonGBChannel channel, ErrorCallback callback){ // 拉流代理通道 try { - StreamInfo streamInfo = streamProxyPlayService.start(channel.getDataDeviceId()); - if (streamInfo == null) { - callback.run(Response.BUSY_HERE, "busy here", null); - }else { - callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); - } + streamProxyPlayService.start(channel.getDataDeviceId(), callback); }catch (Exception e) { callback.run(Response.BUSY_HERE, "busy here", null); } diff --git a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java index 9f9ebf6f..499d2293 100755 --- a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java +++ b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java @@ -1,11 +1,12 @@ package com.genersoft.iot.vmp.streamProxy.service; import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.service.bean.ErrorCallback; import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy; public interface IStreamProxyPlayService { - StreamInfo start(int id); + void start(int id, ErrorCallback callback); StreamInfo startProxy(StreamProxy streamProxy); diff --git a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java index 4cdbcd33..fb191ec3 100755 --- a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java @@ -3,15 +3,27 @@ package com.genersoft.iot.vmp.streamProxy.service.impl; import com.baomidou.dynamic.datasource.annotation.DS; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.exception.ControllerException; +import com.genersoft.iot.vmp.media.bean.MediaInfo; import com.genersoft.iot.vmp.media.bean.MediaServer; +import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent; import com.genersoft.iot.vmp.media.service.IMediaServerService; +import com.genersoft.iot.vmp.service.bean.ErrorCallback; +import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy; import com.genersoft.iot.vmp.streamProxy.dao.StreamProxyMapper; import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyPlayService; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import lombok.extern.slf4j.Slf4j; + +import java.util.concurrent.ConcurrentHashMap; + +import javax.sip.message.Response; + import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.event.EventListener; +import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -29,13 +41,52 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService { @Autowired private IMediaServerService mediaServerService; + private ConcurrentHashMap> callbackMap = new ConcurrentHashMap<>(); + + private ConcurrentHashMap streamInfoMap = new ConcurrentHashMap<>(); + + /** + * 流到来的处理 + */ + @Async("taskExecutor") + @Transactional + @EventListener + public void onApplicationEvent(MediaArrivalEvent event) { + if ("rtsp".equals(event.getSchema())) { + StreamProxy streamProxy = streamProxyMapper.selectOneByAppAndStream(event.getApp(), event.getStream()); + if (streamProxy != null) { + ErrorCallback callback = callbackMap.remove(streamProxy.getId()); + StreamInfo streamInfo = streamInfoMap.remove(streamProxy.getId()); + if (callback != null && streamInfo != null) { + callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + } + } + } + } + @Override - public StreamInfo start(int id) { + public void start(int id, ErrorCallback callback) { StreamProxy streamProxy = streamProxyMapper.select(id); if (streamProxy == null) { throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到"); } - return startProxy(streamProxy); + StreamInfo streamInfo = startProxy(streamProxy); + if (streamInfo == null) { + callback.run(Response.BUSY_HERE, "busy here", null); + return; + } + callbackMap.put(id, callback); + streamInfoMap.put(id, streamInfo); + + MediaServer mediaServer = mediaServerService.getOne(streamProxy.getMediaServerId()); + if (mediaServer != null) { + MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, streamProxy.getApp(), streamProxy.getStream()); + if (mediaInfo != null) { + callbackMap.remove(id); + streamInfoMap.remove(id); + callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo); + } + } } @Override From e370cf880d6fe8ae68feff829950585db9b1000b Mon Sep 17 00:00:00 2001 From: Li Keqing Date: Fri, 3 Jan 2025 21:40:35 +0800 Subject: [PATCH 2/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E7=BD=91=E9=A1=B5?= =?UTF-8?q?=E8=A7=82=E7=9C=8B=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../vmp/streamProxy/service/IStreamProxyPlayService.java | 2 ++ .../service/impl/StreamProxyPlayServiceImpl.java | 9 +++++++++ 2 files changed, 11 insertions(+) diff --git a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java index 499d2293..18b1499a 100755 --- a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java +++ b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java @@ -6,6 +6,8 @@ import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy; public interface IStreamProxyPlayService { + StreamInfo start(int id); + void start(int id, ErrorCallback callback); StreamInfo startProxy(StreamProxy streamProxy); diff --git a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java index fb191ec3..9e5fa140 100755 --- a/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java @@ -89,6 +89,15 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService { } } + @Override + public StreamInfo start(int id) { + StreamProxy streamProxy = streamProxyMapper.select(id); + if (streamProxy == null) { + throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到"); + } + return startProxy(streamProxy); + } + @Override public StreamInfo startProxy(StreamProxy streamProxy){ if (!streamProxy.isEnable()) { From e139676151ccd7b1cbe6b6920f03de2a3636251f Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Sat, 11 Jan 2025 22:32:57 +0800 Subject: [PATCH 3/4] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=9B=A0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E8=B7=AF=E5=BE=84=E5=AE=9A=E4=B9=89=E9=95=BF?= =?UTF-8?q?=E5=BA=A6=E7=9F=AD=E5=AF=BC=E8=87=B4=E8=AE=B0=E5=BD=95=E6=97=A0?= =?UTF-8?q?=E6=B3=95=E6=B7=BB=E5=8A=A0=E7=9A=84BUG?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 数据库/2.7.3/初始化-mysql-2.7.3.sql | 7 +++---- 数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql | 4 ++-- 数据库/2.7.3/更新-mysql-2.7.3.sql | 9 ++++++++- 数据库/2.7.3/更新-postgresql-kingbase-2.7.3.sql | 13 +++++++++++++ 4 files changed, 26 insertions(+), 7 deletions(-) diff --git a/数据库/2.7.3/初始化-mysql-2.7.3.sql b/数据库/2.7.3/初始化-mysql-2.7.3.sql index 17a73f32..c147fd9c 100644 --- a/数据库/2.7.3/初始化-mysql-2.7.3.sql +++ b/数据库/2.7.3/初始化-mysql-2.7.3.sql @@ -336,12 +336,11 @@ create table wvp_cloud_record end_time bigint, media_server_id character varying(50), file_name character varying(255), - folder character varying(255), - file_path character varying(255), + folder character varying(500), + file_path character varying(500), collect bool default false, file_size bigint, - time_len bigint, - constraint uk_stream_push_app_stream_path unique (app, stream, file_path) + time_len bigint ); create table wvp_user diff --git a/数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql b/数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql index 5a2b7fc8..0695e410 100644 --- a/数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql +++ b/数据库/2.7.3/初始化-postgresql-kingbase-2.7.3.sql @@ -352,8 +352,8 @@ create table wvp_cloud_record end_time int8, media_server_id character varying(50), file_name character varying(255), - folder character varying(255), - file_path character varying(255), + folder character varying(500), + file_path character varying(500), collect bool default false, file_size int8, time_len int8, diff --git a/数据库/2.7.3/更新-mysql-2.7.3.sql b/数据库/2.7.3/更新-mysql-2.7.3.sql index 17101e6b..01f42da4 100644 --- a/数据库/2.7.3/更新-mysql-2.7.3.sql +++ b/数据库/2.7.3/更新-mysql-2.7.3.sql @@ -33,4 +33,11 @@ alter table wvp_device_channel drop stream_proxy_id; /* * 20241231 */ -alter table wvp_stream_proxy add relates_media_server_id character varying(50); \ No newline at end of file +alter table wvp_stream_proxy add relates_media_server_id character varying(50); + +/* +* 20250111 +*/ +drop index uk_stream_push_app_stream_path on wvp_cloud_record; +alter table wvp_cloud_record change folder folder varchar(500) null; +alter table wvp_cloud_record change file_path file_path varchar(500) null; \ No newline at end of file diff --git a/数据库/2.7.3/更新-postgresql-kingbase-2.7.3.sql b/数据库/2.7.3/更新-postgresql-kingbase-2.7.3.sql index 237c9d75..8317bce7 100644 --- a/数据库/2.7.3/更新-postgresql-kingbase-2.7.3.sql +++ b/数据库/2.7.3/更新-postgresql-kingbase-2.7.3.sql @@ -27,3 +27,16 @@ set data_type = 3, data_device_id = (SELECT stream_proxy_id from wvp_device_chan alter table wvp_device_channel drop device_db_id; alter table wvp_device_channel drop stream_push_id; alter table wvp_device_channel drop stream_proxy_id; + +/* +* 20241231 +*/ +alter table wvp_stream_proxy add relates_media_server_id character varying(50); + + +/* +* 20250111 +*/ +drop index uk_stream_push_app_stream_path on wvp_cloud_record; +alter table wvp_cloud_record change folder folder varchar(500) null; +alter table wvp_cloud_record change file_path file_path varchar(500) null; \ No newline at end of file From 464c9899862d6437c9df802e836465742acbb5a5 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Sun, 12 Jan 2025 22:24:47 +0800 Subject: [PATCH 4/4] =?UTF-8?q?=E5=9B=BD=E6=A0=87=E6=94=B6=E6=B5=81?= =?UTF-8?q?=E6=97=B6=E6=B5=81=E4=BC=A0=E8=BE=93=E6=A8=A1=E5=BC=8F=E9=BB=98?= =?UTF-8?q?=E8=AE=A4=E4=BD=BF=E7=94=A8TCP=E8=A2=AB=E5=8A=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/gb28181/service/impl/DeviceServiceImpl.java | 2 +- .../transmit/event/request/impl/RegisterRequestProcessor.java | 4 ++-- .../response/cmd/DeviceInfoResponseMessageHandler.java | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java index 410d5a5d..17dadef1 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/DeviceServiceImpl.java @@ -443,7 +443,7 @@ public class DeviceServiceImpl implements IDeviceService { device.setCreateTime(DateUtil.getNow()); device.setUpdateTime(DateUtil.getNow()); if(device.getStreamMode() == null) { - device.setStreamMode("UDP"); + device.setStreamMode("TCP-PASSIVE"); } deviceMapper.addCustomDevice(device); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java index 44a3eb1e..05839f8a 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java @@ -166,7 +166,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen if (device == null) { device = new Device(); - device.setStreamMode("UDP"); + device.setStreamMode("TCP-PASSIVE"); device.setCharset("GB2312"); device.setGeoCoordSys("WGS84"); device.setMediaServerId("auto"); @@ -174,7 +174,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen device.setOnLine(false); } else { if (ObjectUtils.isEmpty(device.getStreamMode())) { - device.setStreamMode("UDP"); + device.setStreamMode("TCP-PASSIVE"); } if (ObjectUtils.isEmpty(device.getCharset())) { device.setCharset("GB2312"); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java index 9db37b41..1de3ff7a 100755 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceInfoResponseMessageHandler.java @@ -79,7 +79,7 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent device.setModel(getText(rootElement, "Model")); device.setFirmware(getText(rootElement, "Firmware")); if (ObjectUtils.isEmpty(device.getStreamMode())) { - device.setStreamMode("UDP"); + device.setStreamMode("TCP-PASSIVE"); } deviceService.updateDevice(device);