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