XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

适配推流列表信息

Browse Source
This commit is contained in:
648540858
2024-07-16 17:33:18 +08:00
parent 90d298f7a7
commit d7abe7fb75
22 changed files with 1154 additions and 884 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/com/genersoft/iot/vmp/gb28181/controller/CommonChannelController.java
View File

@@ -78,4 +78,11 @@ public class CommonChannelController {
public void reset(Integer id){
channelService.reset(id);
}
@Operation(summary = "增加通道", security = @SecurityRequirement(name = JwtUtils.HEADER))
@ResponseBody
@PostMapping("/add")
public void add(@RequestBody CommonGBChannel channel){
channelService.add(channel);
}
}

79
src/main/java/com/genersoft/iot/vmp/gb28181/dao/CommonGBChannelMapper.java
View File

@@ -67,11 +67,6 @@ public interface CommonGBChannelMapper {
" <if test='streamPushId != null' > stream_push_id,</if>" +
"create_time," +
"update_time," +
"sub_count," +
"stream_id," +
"has_audio," +
"gps_time," +
"stream_identification," +
"gb_name," +
"gb_manufacturer," +
"gb_model," +
@@ -111,11 +106,6 @@ public interface CommonGBChannelMapper {
" <if test='streamPushId != null' > #{streamPushId},</if>" +
"#{createTime}, " +
"#{updateTime}, " +
"#{subCount}, " +
"#{streamId}, " +
"#{hasAudio}, " +
"#{gpsTime}, " +
"#{streamIdentification}, " +
"#{gbName}, " +
"#{gbManufacturer}, " +
"#{gbModel}, " +
@@ -351,11 +341,6 @@ public interface CommonGBChannelMapper {
"stream_push_id," +
"create_time," +
"update_time," +
"sub_count," +
"stream_id," +
"has_audio," +
"gps_time," +
"stream_identification," +
"gb_name," +
"gb_manufacturer," +
"gb_model," +
@@ -391,8 +376,8 @@ public interface CommonGBChannelMapper {
"gb_svc_time_support_mode ) " +
"VALUES" +
"<foreach collection='commonGBChannels' index='index' item='item' separator=','> " +
"(#{item.gbDeviceId}, #{item.streamProxyId}, #{item.streamPushId},#{item.createTime},#{item.updateTime},#{item.subCount}," +
"#{item.streamId},#{item.hasAudio},#{item.gpsTime},#{item.streamIdentification},#{item.gbName},#{item.gbManufacturer}, #{item.gbModel}," +
"(#{item.gbDeviceId}, #{item.streamProxyId}, #{item.streamPushId},#{item.createTime},#{item.updateTime}," +
"#{item.gbName},#{item.gbManufacturer}, #{item.gbModel}," +
"#{item.gbOwner},#{item.gbCivilCode},#{item.gbBlock}, #{item.gbAddress}, #{item.gbParental}, #{item.gbParentId},#{item.gbSafetyWay}, " +
"#{item.gbRegisterWay},#{item.gbCertNum},#{item.gbCertifiable},#{item.gbErrCode},#{item.gbEndTime}, #{item.gbSecrecy},#{item.gbIpAddress}," +
"#{item.gbPort},#{item.gbPassword},#{item.gbStatus},#{item.gbLongitude}, #{item.gbLatitude},#{item.gbPtzType},#{item.gbPositionType},#{item.gbRoomType}," +
@@ -473,4 +458,64 @@ public interface CommonGBChannelMapper {
" </script>"})
void reset(@Param("id") int id, @Param("gbDeviceDbId") int gbDeviceDbId, @Param("updateTime") String updateTime);
@Select(value = {" <script>" +
" select\n" +
" id as gb_id,\n" +
" device_db_id,\n" +
" stream_push_id,\n" +
" stream_proxy_id,\n" +
" create_time,\n" +
" update_time,\n" +
" sub_count,\n" +
" stream_id,\n" +
" has_audio,\n" +
" gps_time,\n" +
" stream_identification,\n" +
" coalesce(gb_device_id, device_id) as gb_device_id,\n" +
" coalesce(gb_name, name) as gb_name,\n" +
" coalesce(gb_manufacturer, manufacturer) as gb_manufacturer,\n" +
" coalesce(gb_model, model) as gb_model,\n" +
" coalesce(gb_owner, owner) as gb_owner,\n" +
" coalesce(gb_civil_code, civil_code) as gb_civil_code,\n" +
" coalesce(gb_block, block) as gb_block,\n" +
" coalesce(gb_address, address) as gb_address,\n" +
" coalesce(gb_parental, parental) as gb_parental,\n" +
" coalesce(gb_parent_id, parent_id) as gb_parent_id,\n" +
" coalesce(gb_safety_way, safety_way) as gb_safety_way,\n" +
" coalesce(gb_register_way, register_way) as gb_register_way,\n" +
" coalesce(gb_cert_num, cert_num) as gb_cert_num,\n" +
" coalesce(gb_certifiable, certifiable) as gb_certifiable,\n" +
" coalesce(gb_err_code, err_code) as gb_err_code,\n" +
" coalesce(gb_end_time, end_time) as gb_end_time,\n" +
" coalesce(gb_secrecy, secrecy) as gb_secrecy,\n" +
" coalesce(gb_ip_address, ip_address) as gb_ip_address,\n" +
" coalesce(gb_port, port) as gb_port,\n" +
" coalesce(gb_password, password) as gb_password,\n" +
" coalesce(gb_status, status) as gb_status,\n" +
" coalesce(gb_longitude, longitude) as gb_longitude,\n" +
" coalesce(gb_latitude, latitude) as gb_latitude,\n" +
" coalesce(gb_ptz_type, ptz_type) as gb_ptz_type,\n" +
" coalesce(gb_position_type, position_type) as gb_position_type,\n" +
" coalesce(gb_room_type, room_type) as gb_room_type,\n" +
" coalesce(gb_use_type, use_type) as gb_use_type,\n" +
" coalesce(gb_supply_light_type, supply_light_type) as gb_supply_light_type,\n" +
" coalesce(gb_direction_type, direction_type) as gb_direction_type,\n" +
" coalesce(gb_resolution, resolution) as gb_resolution,\n" +
" coalesce(gb_business_group_id, business_group_id) as gb_business_group_id,\n" +
" coalesce(gb_download_speed, download_speed) as gb_download_speed,\n" +
" coalesce(gb_svc_space_support_mod, svc_space_support_mod) as gb_svc_space_support_mod,\n" +
" coalesce(gb_svc_time_support_mode,svc_time_support_mode) as gb_svc_time_support_mode\n" +
" from wvp_device_channel" +
" where 1 = 1 and id in " +
" <foreach collection='commonGBChannelList' item='item' open='(' separator=',' close=')' > #{item.gbId}</foreach>" +
"</script>"})
List<CommonGBChannel> queryByIds(List<CommonGBChannel> commonGBChannelList);
@Delete(value = {" <script>" +
" delete from wvp_device_channel" +
" where 1 = 1 and id in " +
" <foreach collection='channelListInDb' item='item' open='(' separator=',' close=')' > #{item.gbId}</foreach>" +
"</script>"})
void batchDelete(List<CommonGBChannel> channelListInDb);
}

3
src/main/java/com/genersoft/iot/vmp/gb28181/service/IGbChannelService.java
View File

@@ -15,6 +15,8 @@ public interface IGbChannelService {
int delete(int gbId);
void delete(List<CommonGBChannel> commonGBChannelList);
int update(CommonGBChannel commonGBChannel);
int offline(CommonGBChannel commonGBChannel);
@@ -40,4 +42,5 @@ public interface IGbChannelService {
List<NetworkIdentificationType> getNetworkIdentificationTypeList();
void reset(int id);
}

17
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelServiceImpl.java
View File

@@ -52,9 +52,24 @@ public class GbChannelServiceImpl implements IGbChannelService {
return 1;
}
@Override
public void delete(List<CommonGBChannel> commonGBChannelList) {
List<CommonGBChannel> channelListInDb = commonGBChannelMapper.queryByIds(commonGBChannelList);
if (channelListInDb.isEmpty()) {
return;
}
commonGBChannelMapper.batchDelete(channelListInDb);
try {
// 发送通知
eventPublisher.catalogEventPublish(null, channelListInDb, CatalogEvent.DEL);
}catch (Exception e) {
log.warn("[通道移除通知] 发送失败，{}条", channelListInDb.size(), e);
}
}
@Override
public int update(CommonGBChannel commonGBChannel) {
log.warn("[更新通道] 通道ID: {}, ", commonGBChannel.getGbId());
log.info("[更新通道] 通道ID: {}, ", commonGBChannel.getGbId());
if (commonGBChannel.getGbId() <= 0) {
log.warn("[更新通道] 未找到数据库ID，更新失败， {}", commonGBChannel.getGbDeviceDbId());
return 0;

2
src/main/java/com/genersoft/iot/vmp/streamProxy/bean/StreamProxy.java
View File

@@ -71,7 +71,7 @@ public class StreamProxy extends CommonGBChannel {
@Schema(description = "创建时间")
private String createTime;
public CommonGBChannel getCommonGBChannel() {
public CommonGBChannel buildCommonGBChannel() {
if (ObjectUtils.isEmpty(this.getGbDeviceId())) {
return null;
}

16
src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyServiceImpl.java
View File

@@ -169,7 +169,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
streamProxy.setUpdateTime(now);
if (streamProxyMapper.add(streamProxy) > 0 && !ObjectUtils.isEmpty(streamProxy.getGbDeviceId())) {
gbChannelService.add(streamProxy.getCommonGBChannel());
gbChannelService.add(streamProxy.buildCommonGBChannel());
}
return true;
}
@@ -183,9 +183,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
if (streamProxyMapper.update(streamProxy) > 0 && !ObjectUtils.isEmpty(streamProxy.getGbDeviceId())) {
if (streamProxy.getGbId() > 0) {
gbChannelService.update(streamProxy.getCommonGBChannel());
gbChannelService.update(streamProxy.buildCommonGBChannel());
}else {
gbChannelService.add(streamProxy.getCommonGBChannel());
gbChannelService.add(streamProxy.buildCommonGBChannel());
}
}
return true;
@@ -303,7 +303,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
redisCatchStorage.addStream(mediaServer, "pull", streamInfo.getApp(), streamInfo.getStream(), streamInfo.getMediaInfo());
if ("OFF".equalsIgnoreCase(streamProxy.getGbStatus()) && streamProxy.getGbId() > 0) {
streamProxy.setGbStatus("ON");
channelListForOnline.add(streamProxy.getCommonGBChannel());
channelListForOnline.add(streamProxy.buildCommonGBChannel());
}
streamProxyMapForDb.remove(key);
}
@@ -319,7 +319,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
for (StreamProxy streamProxy : streamProxyMapForDb.values()) {
if ("ON".equalsIgnoreCase(streamProxy.getGbStatus()) && streamProxy.getGbId() > 0) {
streamProxy.setGbStatus("OFF");
channelListForOffline.add(streamProxy.getCommonGBChannel());
channelListForOffline.add(streamProxy.buildCommonGBChannel());
}
// 移除开启了无人观看自动移除的流
if (streamProxy.getGbDeviceId() == null && streamProxy.isEnableRemoveNoneReader()) {
@@ -360,7 +360,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
for (StreamProxy streamProxy : streamProxies) {
if (streamProxy.getGbId() > 0 && "ON".equalsIgnoreCase(streamProxy.getGbStatus())) {
channelListForOffline.add(streamProxy.getCommonGBChannel());
channelListForOffline.add(streamProxy.buildCommonGBChannel());
}
if (streamProxy.getGbId() == 0 && streamProxy.isEnableRemoveNoneReader()) {
streamProxiesForRemove.add(streamProxy);
@@ -400,9 +400,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
streamProxy.setGbStatus(status?"ON":"OFF");
if (streamProxy.getGbId() > 0) {
if (status) {
gbChannelService.online(streamProxy.getCommonGBChannel());
gbChannelService.online(streamProxy.buildCommonGBChannel());
}else {
gbChannelService.offline(streamProxy.getCommonGBChannel());
gbChannelService.offline(streamProxy.buildCommonGBChannel());
}
}

10
src/main/java/com/genersoft/iot/vmp/streamPush/bean/BatchRemoveParam.java Normal file
View File

@@ -0,0 +1,10 @@
package com.genersoft.iot.vmp.streamPush.bean;
import lombok.Data;
import java.util.Set;
@Data
public class BatchRemoveParam {
private Set<Integer> ids;
}

16
src/main/java/com/genersoft/iot/vmp/streamPush/bean/StreamPush.java
View File

@@ -83,13 +83,13 @@ public class StreamPush extends CommonGBChannel implements Comparable<StreamPush
}
public StreamPush getInstance(StreamInfo streamInfo) {
StreamPush streamPushItem = new StreamPush();
streamPushItem.setApp(streamInfo.getApp());
streamPushItem.setMediaServerId(streamInfo.getMediaServerId());
streamPushItem.setStream(streamInfo.getStream());
streamPushItem.setCreateTime(DateUtil.getNow());
streamPushItem.setServerId(streamInfo.getMediaServerId());
return streamPushItem;
StreamPush streamPush = new StreamPush();
streamPush.setApp(streamInfo.getApp());
streamPush.setMediaServerId(streamInfo.getMediaServerId());
streamPush.setStream(streamInfo.getStream());
streamPush.setCreateTime(DateUtil.getNow());
streamPush.setServerId(streamInfo.getMediaServerId());
return streamPush;
}
@@ -103,7 +103,7 @@ public class StreamPush extends CommonGBChannel implements Comparable<StreamPush
return streamPushItem;
}
public CommonGBChannel getCommonGBChannel() {
public CommonGBChannel buildCommonGBChannel() {
if (ObjectUtils.isEmpty(this.getGbDeviceId())) {
return null;
}

38
src/main/java/com/genersoft/iot/vmp/streamPush/controller/StreamPushController.java
View File

@@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.streamPush.bean.BatchRemoveParam;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.streamPush.bean.StreamPushExcelDto;
import com.genersoft.iot.vmp.streamPush.enent.StreamPushUploadFileHandler;
@@ -88,13 +89,12 @@ public class StreamPushController {
}
@PostMapping(value = "/stop")
@PostMapping(value = "/remove")
@ResponseBody
@Operation(summary = "中止一个推流", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "app", description = "应用名", required = true)
@Parameter(name = "stream", description = "流id", required = true)
public void stop(String app, String stream){
if (!streamPushService.stopByAppAndStream(app, stream)){
@Operation(summary = "删除", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "id", description = "应用名", required = true)
public void delete(int id){
if (streamPushService.delete(id) > 0){
throw new ControllerException(ErrorCode.ERROR100);
}
}
@@ -232,7 +232,7 @@ public class StreamPushController {
@PostMapping(value = "/add")
@ResponseBody
@Operation(summary = "添加推流信息", security = @SecurityRequirement(name = JwtUtils.HEADER))
public void add(@RequestBody StreamPush stream){
public StreamPush add(@RequestBody StreamPush stream){
if (ObjectUtils.isEmpty(stream.getGbId())) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "国标ID不可为空");
}
@@ -244,5 +244,29 @@ public class StreamPushController {
if (!streamPushService.add(stream)) {
throw new ControllerException(ErrorCode.ERROR100);
}
stream.setStreamPushId(stream.getId());
return stream;
}
@PostMapping(value = "/update")
@ResponseBody
@Operation(summary = "更新推流信息", security = @SecurityRequirement(name = JwtUtils.HEADER))
public void update(@RequestBody StreamPush stream){
if (ObjectUtils.isEmpty(stream.getId())) {
throw new ControllerException(ErrorCode.ERROR400.getCode(), "ID不可为空");
}
if (!streamPushService.update(stream)) {
throw new ControllerException(ErrorCode.ERROR100);
}
}
@DeleteMapping(value = "/batchRemove")
@ResponseBody
@Operation(summary = "删除多个推流", security = @SecurityRequirement(name = JwtUtils.HEADER))
public void batchStop(@RequestBody BatchRemoveParam ids){
if(ids.getIds().isEmpty()) {
return;
}
streamPushService.batchRemove(ids.getIds());
}
}

208
src/main/java/com/genersoft/iot/vmp/streamPush/dao/StreamPushMapper.java
View File

@@ -1,6 +1,5 @@
package com.genersoft.iot.vmp.streamPush.dao;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import org.apache.ibatis.annotations.*;
@@ -8,207 +7,132 @@ import org.springframework.stereotype.Repository;
import java.util.List;
import java.util.Map;
import java.util.Set;
@Mapper
@Repository
public interface StreamPushMapper {
@Insert("INSERT INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
"push_time, alive_second, media_server_id, server_id, update_time, create_time, push_ing, self) VALUES" +
"(#{app}, #{stream}, #{totalReaderCount}, #{originType}, #{originTypeStr}, " +
"#{pushTime}, #{aliveSecond}, #{mediaServerId} , #{serverId} , #{updateTime} , #{createTime}, " +
"#{pushIng}, #{self} )")
@Insert("INSERT INTO wvp_stream_push (app, stream, media_server_id, server_id, push_time, update_time, create_time, push_ing) VALUES" +
"(#{app}, #{stream}, #{mediaServerId} , #{serverId} , #{pushTime} ,#{updateTime}, #{createTime}, #{pushIng})")
@Options(useGeneratedKeys = true, keyProperty = "id", keyColumn = "id")
int add(StreamPush streamPushItem);
@Update(value = {" <script>" +
"UPDATE wvp_stream_push " +
"SET update_time=#{updateTime}" +
"<if test=\"app != null\">, app=#{app}</if>" +
"<if test=\"stream != null\">, stream=#{stream}</if>" +
"<if test=\"mediaServerId != null\">, media_server_id=#{mediaServerId}</if>" +
"<if test=\"serverId != null\">, server_id=#{serverId}</if>" +
"<if test=\"totalReaderCount != null\">, total_reader_count=#{totalReaderCount}</if>" +
"<if test=\"originType != null\">, origin_type=#{originType}</if>" +
"<if test=\"originTypeStr != null\">, origin_type_str=#{originTypeStr}</if>" +
"<if test=\"pushTime != null\">, push_time=#{pushTime}</if>" +
"<if test=\"aliveSecond != null\">, alive_second=#{aliveSecond}</if>" +
"<if test=\"pushIng != null\">, push_ing=#{pushIng}</if>" +
"<if test=\"self != null\">, self=#{self}</if>" +
"WHERE app=#{app} AND stream=#{stream}"+
"WHERE id = #{id}"+
" </script>"})
int update(StreamPush streamPushItem);
@Delete("DELETE FROM wvp_stream_push WHERE app=#{app} AND stream=#{stream}")
int del(String app, String stream);
@Delete("<script> "+
"DELETE sp FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream where " +
"<foreach collection='streamPushItems' item='item' separator='or'>" +
"(sp.app=#{item.app} and sp.stream=#{item.stream} and gs.gb_id is null) " +
"</foreach>" +
"</script>")
int delAllWithoutGBId(List<StreamPush> streamPushItems);
@Delete("<script> "+
"DELETE FROM wvp_stream_push where " +
"<foreach collection='streamPushItems' item='item' separator='or'>" +
"(app=#{item.app} and stream=#{item.stream}) " +
"</foreach>" +
"</script>")
int delAll(List<StreamPush> streamPushItems);
@Delete("<script> "+
"DELETE FROM wvp_stream_push where " +
"<foreach collection='gbStreams' item='item' separator='or'>" +
"(app=#{item.app} and stream=#{item.stream}) " +
"</foreach>" +
"</script>")
int delAllForGbStream(List<GbStream> gbStreams);
@Delete("DELETE FROM wvp_stream_push WHERE id=#{id}")
int del(@Param("id") int id);
@Select(value = {" <script>" +
"SELECT " +
"st.*, " +
"gs.gb_id, gs.name, gs.longitude, gs.latitude, gs.gb_stream_id " +
"from " +
"wvp_stream_push st " +
"LEFT join wvp_gb_stream gs " +
"on st.app = gs.app AND st.stream = gs.stream " +
"WHERE " +
"1=1 " +
" <if test='query != null'> AND (st.app LIKE concat('%',#{query},'%') OR st.stream LIKE concat('%',#{query},'%') OR gs.gb_id LIKE concat('%',#{query},'%') OR gs.name LIKE concat('%',#{query},'%'))</if> " +
" <if test='pushing == true' > AND (gs.gb_id is null OR st.push_ing=1)</if>" +
" <if test='pushing == false' > AND (st.push_ing is null OR st.push_ing=0) </if>" +
" SELECT " +
" st.*, " +
" st.id as stream_push_id, " +
" wdc.*, " +
" wdc.id as gb_id" +
" from " +
" wvp_stream_push st " +
" LEFT join wvp_device_channel wdc " +
" on st.id = wdc.stream_push_id " +
" WHERE " +
" 1=1 " +
" <if test='query != null'> AND (st.app LIKE concat('%',#{query},'%') OR st.stream LIKE concat('%',#{query},'%') " +
" OR wdc.gb_device_id LIKE concat('%',#{query},'%') OR wdc.gb_name LIKE concat('%',#{query},'%'))</if> " +
" <if test='pushing == true' > AND st.push_ing=1</if>" +
" <if test='pushing == false' > AND st.push_ing=0 </if>" +
" <if test='mediaServerId != null' > AND st.media_server_id=#{mediaServerId} </if>" +
"order by st.create_time desc" +
" order by st.create_time desc" +
" </script>"})
List<StreamPush> selectAllForList(@Param("query") String query, @Param("pushing") Boolean pushing, @Param("mediaServerId") String mediaServerId);
@Select("SELECT st.*, gs.gb_id, gs.name, gs.longitude, gs.latitude FROM wvp_stream_push st LEFT join wvp_gb_stream gs on st.app = gs.app AND st.stream = gs.stream order by st.create_time desc")
List<StreamPush> selectAll();
@Select("SELECT st.*, gs.gb_id, gs.name, gs.longitude, gs.latitude FROM wvp_stream_push st LEFT join wvp_gb_stream gs on st.app = gs.app AND st.stream = gs.stream WHERE st.app=#{app} AND st.stream=#{stream}")
@Select("SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id WHERE st.app=#{app} AND st.stream=#{stream}")
StreamPush selectByAppAndStream(@Param("app") String app, @Param("stream") String stream);
@Insert("<script>" +
"Insert INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
"create_time, alive_second, media_server_id, server_id, status, push_ing) " +
"VALUES <foreach collection='streamPushItems' item='item' index='index' separator=','>" +
"( #{item.app}, #{item.stream}, #{item.totalReaderCount}, #{item.originType}, " +
"#{item.originTypeStr},#{item.createTime}, #{item.aliveSecond}, #{item.mediaServerId},#{item.serverId}, #{item.status} ," +
" #{item.pushIng} )" +
"Insert INTO wvp_stream_push ( " +
" app, stream, media_server_id, server_id, push_time, update_time, create_time, push_ing) " +
" VALUES <foreach collection='streamPushItems' item='item' index='index' separator=','>" +
" ( #{item.app}, #{item.stream}, #{item.mediaServerId},#{item.serverId} ,#{item.pushTime}, #{item.updateTime}, #{item.createTime}, #{item.pushIng} )" +
" </foreach>" +
"</script>")
" </script>")
@Options(useGeneratedKeys = true, keyProperty = "id", keyColumn = "id")
int addAll(List<StreamPush> streamPushItems);
@Delete("DELETE FROM wvp_stream_push")
void clear();
@Delete("delete" +
" from wvp_stream_push " +
" where id in " +
" (select temp.id from " +
" (select wgs.gb_stream_id as id " +
" from wvp_gb_stream wgs" +
" left join wvp_stream_push sp on sp.id = wgs.gb_stream_id" +
" where wgs.gb_id is null and wgs.media_server_id = #{mediaServerId}) temp)"
)
void deleteWithoutGBId(String mediaServerId);
@Select("SELECT * FROM wvp_stream_push WHERE media_server_id=#{mediaServerId}")
@Select("SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id WHERE st.media_server_id=#{mediaServerId}")
List<StreamPush> selectAllByMediaServerId(String mediaServerId);
@Select("SELECT sp.* FROM wvp_stream_push sp left join wvp_gb_stream gs on gs.app = sp.app and gs.stream= sp.stream WHERE sp.media_server_id=#{mediaServerId} and gs.gb_id is null")
@Select("SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id WHERE st.media_server_id=#{mediaServerId} and wdc.gb_device_id is null")
List<StreamPush> selectAllByMediaServerIdWithOutGbID(String mediaServerId);
@Update("UPDATE wvp_stream_push " +
"SET status=#{status} " +
"WHERE app=#{app} AND stream=#{stream}")
int updateStatus(@Param("app") String app, @Param("stream") String stream, @Param("status") boolean status);
@Update("UPDATE wvp_stream_push " +
"SET push_ing=#{pushIng} " +
"WHERE app=#{app} AND stream=#{stream}")
int updatePushStatus(@Param("app") String app, @Param("stream") String stream, @Param("pushIng") boolean pushIng);
@Update("UPDATE wvp_stream_push " +
"SET status=#{status} " +
"WHERE media_server_id=#{mediaServerId}")
void updateStatusByMediaServerId(@Param("mediaServerId") String mediaServerId, @Param("status") boolean status);
"WHERE id=#{id}")
int updatePushStatus(@Param("id") int id, @Param("pushIng") boolean pushIng);
@Select("<script> "+
"SELECT gs.* FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " +
"where sp.status = true and (gs.app, gs.stream) in (" +
"SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id " +
"where (st.app, st.stream) in (" +
"<foreach collection='offlineStreams' item='item' separator=','>" +
"(#{item.app}, #{item.stream}) " +
"</foreach>" +
")</script>")
List<StreamPush> getOnlinePusherForGbInList(List<StreamPushItemFromRedis> offlineStreams);
List<StreamPush> getListFromRedis(List<StreamPushItemFromRedis> offlineStreams);
@Update("<script> "+
"UPDATE wvp_stream_push SET status=0 where (app, stream) in (" +
"<foreach collection='offlineStreams' item='item' separator=','>" +
"(#{item.app}, #{item.stream}) " +
"</foreach>" +
")</script>")
void offline(List<StreamPushItemFromRedis> offlineStreams);
@Select("<script> "+
"SELECT * FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " +
"where sp.status = 0 and (gs.app, gs.stream) in (" +
"<foreach collection='onlineStreams' item='item' separator=','>" +
"(#{item.app}, #{item.stream}) " +
"</foreach>" +
") </script>")
List<StreamPush> getOfflinePusherForGbInList(List<StreamPushItemFromRedis> onlineStreams);
@Update("<script> "+
"UPDATE wvp_stream_push SET status=1 where (app, stream) in (" +
"<foreach collection='onlineStreams' item='item' separator=','>" +
"(#{item.app}, #{item.stream}) " +
"</foreach>" +
")</script>")
void online(List<StreamPushItemFromRedis> onlineStreams);
@Select("SELECT gs.* FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream where sp.status = true")
List<StreamPush> getOnlinePusherForGb();
@Update("UPDATE wvp_stream_push SET status=0")
void setAllStreamOffline();
@Select("SELECT CONCAT(app,stream) from wvp_gb_stream")
@Select("SELECT CONCAT(app,stream) from wvp_stream_push")
List<String> getAllAppAndStream();
@Select("select count(1) from wvp_stream_push ")
int getAllCount();
@Select(value = {" <script>" +
" <if test='pushIngAsOnline == true'> select count(1) from wvp_stream_push where push_ing = true </if>" +
" <if test='pushIngAsOnline == false'> select count(1)from wvp_stream_push where status = true </if>" +
" select count(1) from wvp_stream_push where push_ing = true" +
" </script>"})
int getAllOnline(Boolean usePushingAsStatus);
@Select("<script> " +
"select app, stream from wvp_stream_push where (app, stream) in " +
"<foreach collection='streamPushItems' item='item' separator=','>" +
"(#{item.app}, #{item.stream}) " +
"</foreach>" +
"</script>")
List<StreamPush> getListIn(List<StreamPush> streamPushItems);
int getAllPushing(Boolean usePushingAsStatus);
@MapKey("vhost")
@Select("SELECT CONCAT(wsp.app, wsp.stream) as vhost, wsp.app, wsp.stream, wgs.gb_id, wgs.name " +
@Select("SELECT CONCAT(wsp.app, wsp.stream) as vhost, wsp.*, wsp.* , wsp.id as gb_id " +
" from wvp_stream_push wsp " +
" left join wvp_gb_stream wgs on wgs.app = wsp.app and wgs.stream = wsp.stream")
" LEFT join wvp_device_channel wdc on wsp.id = wdc.stream_push_id")
Map<String, StreamPush> getAllAppAndStreamMap();
@MapKey("gb_id")
@Select("SELECT wgs.gb_id, wsp.app, wsp.stream, wgs.gb_id, wgs.name " +
@MapKey("gb_device_id")
@Select("SELECT wdc.gb_device_id, wsp.id as stream_push_id, wsp.*, wsp.* , wsp.id as gb_id " +
" from wvp_stream_push wsp " +
" left join wvp_gb_stream wgs on wgs.app = wsp.app and wgs.stream = wsp.stream")
" LEFT join wvp_device_channel wdc on wsp.id = wdc.stream_push_id")
Map<String, StreamPush> getAllGBId();
StreamPush select(Integer streamPushId);
@Select("SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id WHERE st.id=#{id}")
StreamPush select(@Param("id") int id);
@Select("<script> "+
"SELECT st.*, st.id as stream_push_id, wdc.*, wdc.id as gb_id FROM wvp_stream_push st LEFT join wvp_device_channel wdc on st.id = wdc.stream_push_id " +
" where st.id in (" +
" <foreach collection='ids' item='item' separator=','>" +
" #{item} " +
" </foreach>" +
" )</script>")
List<StreamPush> selectInSet(Set<Integer> ids);
@Delete("<script> "+
"DELETE FROM wvp_stream_push WHERE" +
" id in (" +
"<foreach collection='streamPushList' item='item' separator=','>" +
" #{item.id} " +
"</foreach>" +
")</script>")
void batchDel(List<StreamPush> streamPushList);
}

2
src/main/java/com/genersoft/iot/vmp/streamPush/enent/StreamPushUploadFileHandler.java
View File

@@ -110,7 +110,7 @@ public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPus
streamPush.setGbName(streamPushExcelDto.getName());
streamPush.setGbLongitude(streamPushExcelDto.getLongitude());
streamPush.setGbLatitude(streamPushExcelDto.getLatitude());
streamPush.setUpdateTime(DateUtil.getNow());
streamPushItemForSave.put(streamPush.getApp() + streamPush.getStream(), streamPush);
loadedSize ++;

5
src/main/java/com/genersoft/iot/vmp/streamPush/service/IStreamPushService.java
View File

@@ -7,6 +7,7 @@ import com.github.pagehelper.PageInfo;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* @author lin
@@ -92,4 +93,8 @@ public interface IStreamPushService {
void updatePushStatus(Integer streamPushId, boolean pushIng);
void batchUpdate(List<StreamPush> streamPushItemForUpdate);
int delete(int id);
void batchRemove(Set<Integer> ids);
}

84
src/main/java/com/genersoft/iot/vmp/streamPush/service/impl/StreamPushServiceImpl.java
View File

@@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONObject;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
import com.genersoft.iot.vmp.media.bean.MediaInfo;
@@ -21,6 +22,7 @@ import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
import com.genersoft.iot.vmp.streamPush.dao.StreamPushMapper;
import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.github.pagehelper.PageHelper;
@@ -205,6 +207,10 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Transactional
public boolean add(StreamPush stream) {
log.info("[添加推流] app: {}, stream: {}, 国标编号: {}", stream.getApp(), stream.getStream(), stream.getGbDeviceId());
StreamPush streamPushInDb = streamPushMapper.selectByAppAndStream(stream.getApp(), stream.getStream());
if (streamPushInDb != null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "应用名+流ID已存在");
}
stream.setUpdateTime(DateUtil.getNow());
stream.setCreateTime(DateUtil.getNow());
int addResult = streamPushMapper.add(stream);
@@ -218,7 +224,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
if (channel != null) {
log.info("[添加推流]失败，国标编号已存在: {} app: {}, stream: {}, ", stream.getGbDeviceId(), stream.getApp(), stream.getStream());
}
int addChannelResult = gbChannelService.add(stream.getCommonGBChannel());
int addChannelResult = gbChannelService.add(stream.buildCommonGBChannel());
return addChannelResult > 0;
}
@@ -243,10 +249,18 @@ public class StreamPushServiceImpl implements IStreamPushService {
public boolean update(StreamPush streamPush) {
log.info("[更新推流]：id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
assert streamPush.getId() != null;
StreamPush streamPushInDb = streamPushMapper.select(streamPush.getId());
if (!streamPushInDb.getApp().equals(streamPush.getApp()) || !streamPushInDb.getStream().equals(streamPush.getStream())) {
// app或者stream变化
StreamPush streamPushInDbForAppAndStream = streamPushMapper.selectByAppAndStream(streamPush.getApp(), streamPush.getStream());
if (streamPushInDbForAppAndStream != null && !streamPushInDbForAppAndStream.getId().equals(streamPush.getId())) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "应用名+流ID已存在");
}
}
streamPush.setUpdateTime(DateUtil.getNow());
streamPushMapper.update(streamPush);
if (streamPush.getGbId() > 0) {
gbChannelService.update(streamPush.getCommonGBChannel());
gbChannelService.update(streamPush.buildCommonGBChannel());
}
return true;
}
@@ -282,7 +296,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
streamPush.setPushIng(false);
if (userSetting.isUsePushingAsStatus()) {
gbChannelService.offline(streamPush.getCommonGBChannel());
gbChannelService.offline(streamPush.buildCommonGBChannel());
}
redisCatchStorage.deleteSendRTPServer(null, streamPush.getGbDeviceId(), null, streamPush.getStream());
mediaServerService.stopSendRtp(mediaServer, streamPush.getApp(), streamPush.getStream(), null);
@@ -440,8 +454,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
streamPushMapper.addAll(streamPushItems);
List<CommonGBChannel> commonGBChannels = new ArrayList<>();
for (StreamPush streamPush : streamPushItems) {
if (ObjectUtils.isEmpty(streamPush.getGbDeviceId())) {
commonGBChannels.add(streamPush.getCommonGBChannel());
if (!ObjectUtils.isEmpty(streamPush.getGbDeviceId())) {
commonGBChannels.add(streamPush.buildCommonGBChannel());
}
}
gbChannelService.batchAdd(commonGBChannels);
@@ -449,13 +463,13 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public void allOffline() {
List<StreamPush> onlinePushers = streamPushMapper.getOnlinePusherForGb();
if (onlinePushers.isEmpty()) {
List<StreamPush> streamPushList = streamPushMapper.selectAllForList(null, null, null);
if (streamPushList.isEmpty()) {
return;
}
List<CommonGBChannel> commonGBChannelList = new ArrayList<>();
for (StreamPush onlinePusher : onlinePushers) {
commonGBChannelList.add(onlinePusher.getCommonGBChannel());
for (StreamPush streamPush : streamPushList) {
commonGBChannelList.add(streamPush.buildCommonGBChannel());
}
gbChannelService.offline(commonGBChannelList);
}
@@ -463,10 +477,10 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public void offline(List<StreamPushItemFromRedis> offlineStreams) {
// 更新部分设备离线
List<StreamPush> onlinePushers = streamPushMapper.getOnlinePusherForGbInList(offlineStreams);
List<StreamPush> streamPushList = streamPushMapper.getListFromRedis(offlineStreams);
List<CommonGBChannel> commonGBChannelList = new ArrayList<>();
for (StreamPush onlinePusher : onlinePushers) {
commonGBChannelList.add(onlinePusher.getCommonGBChannel());
for (StreamPush onlinePusher : streamPushList) {
commonGBChannelList.add(onlinePusher.buildCommonGBChannel());
}
gbChannelService.offline(commonGBChannelList);
}
@@ -474,10 +488,10 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public void online(List<StreamPushItemFromRedis> onlineStreams) {
// 更新部分设备上线streamPushService
List<StreamPush> offlinePushers = streamPushMapper.getOfflinePusherForGbInList(onlineStreams);
List<StreamPush> streamPushList = streamPushMapper.getListFromRedis(onlineStreams);
List<CommonGBChannel> commonGBChannelList = new ArrayList<>();
for (StreamPush onlinePusher : offlinePushers) {
commonGBChannelList.add(onlinePusher.getCommonGBChannel());
for (StreamPush onlinePusher : streamPushList) {
commonGBChannelList.add(onlinePusher.buildCommonGBChannel());
}
gbChannelService.online(commonGBChannelList);
}
@@ -491,7 +505,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public ResourceBaseInfo getOverview() {
int total = streamPushMapper.getAllCount();
int online = streamPushMapper.getAllOnline(userSetting.isUsePushingAsStatus());
int online = streamPushMapper.getAllPushing(userSetting.isUsePushingAsStatus());
return new ResourceBaseInfo(total, online);
}
@@ -512,9 +526,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
return;
}
if ("ON".equalsIgnoreCase(push.getGbStatus())) {
gbChannelService.online(push.getCommonGBChannel());
gbChannelService.online(push.buildCommonGBChannel());
}else {
gbChannelService.offline(push.getCommonGBChannel());
gbChannelService.offline(push.buildCommonGBChannel());
}
}
@@ -556,4 +570,38 @@ public class StreamPushServiceImpl implements IStreamPushService {
public void batchUpdate(List<StreamPush> streamPushItemForUpdate) {
}
@Override
@Transactional
public int delete(int id) {
StreamPush streamPush = streamPushMapper.select(id);
if (streamPush == null) {
return 0;
}
if(streamPush.isPushIng()) {
MediaServer mediaServer = mediaServerService.getOne(streamPush.getMediaServerId());
mediaServerService.closeStreams(mediaServer, streamPush.getApp(), streamPush.getStream());
}
if (streamPush.getGbDeviceId() != null) {
gbChannelService.delete(streamPush.getGbId());
}
return streamPushMapper.del(id);
}
@Override
@Transactional
public void batchRemove(Set<Integer> ids) {
List<StreamPush> streamPushList = streamPushMapper.selectInSet(ids);
if (streamPushList.isEmpty()) {
return;
}
List<CommonGBChannel> commonGBChannelList = new ArrayList<>();
streamPushList.stream().forEach(streamPush -> {
if (streamPush.getGbDeviceId() != null) {
commonGBChannelList.add(streamPush.buildCommonGBChannel());
}
});
streamPushMapper.batchDel(streamPushList);
gbChannelService.delete(commonGBChannelList);
}
}