Shikong/gb28181-docking-platform-proxy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

调整

Browse Source
This commit is contained in:
shikong 2023-12-14 17:09:00 +08:00
parent a5faf10138
commit 2bb56f5bbc
1 changed files with 21 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
gb28181-wvp-proxy-service/src/main/java/cn/skcks/docking/gb28181/wvp/service/gb28181/Gb28181DownloadService.java
View File

@ -81,6 +81,7 @@ public class Gb28181DownloadService {
private final WvpProxyConfig wvpProxyConfig; private final WvpProxyConfig wvpProxyConfig;
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
private final ConcurrentMap<String, DeferredResult<JsonResponse<String>>> requestMap = new ConcurrentHashMap<>(); private final ConcurrentMap<String, DeferredResult<JsonResponse<String>>> requestMap = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> realtimeVideo = new ConcurrentHashMap<>();
private final ConcurrentMap<String, VideoInfo> realtimeVideoInfoMap = new ConcurrentHashMap<>(); private final ConcurrentMap<String, VideoInfo> realtimeVideoInfoMap = new ConcurrentHashMap<>();
@Getter @Getter
@ -224,7 +225,7 @@ public class Gb28181DownloadService {
} }
@SneakyThrows @SneakyThrows
public DeferredResult<JsonResponse<String>> realtimeVideoUrl(String deviceCode){ public synchronized DeferredResult<JsonResponse<String>> realtimeVideoUrl(String deviceCode){
DeferredResult<JsonResponse<String>> result = new DeferredResult<>(TimeUnit.SECONDS.toMillis(60)); DeferredResult<JsonResponse<String>> result = new DeferredResult<>(TimeUnit.SECONDS.toMillis(60));
result.onTimeout(()->{ result.onTimeout(()->{
result.setResult(JsonResponse.error("请求超时")); result.setResult(JsonResponse.error("请求超时"));
@ -239,9 +240,11 @@ public class Gb28181DownloadService {
result.setResult(JsonResponse.error("设备(通道)不存在")); result.setResult(JsonResponse.error("设备(通道)不存在"));
return result; return result;
} }
closeExistRequest(deviceCode, device, docking); String existUrl = realtimeVideo.get(deviceCode);
requestMap.put(deviceCode, result); if(Optional.ofNullable(existUrl).isPresent()){
result.setResult(JsonResponse.success(existUrl));
return result;
}
// 间隔一定时间(200ms) 给设备足够的时间结束前次请求 // 间隔一定时间(200ms) 给设备足够的时间结束前次请求
scheduledExecutorService.schedule(()->{ scheduledExecutorService.schedule(()->{
realtime.realtime(deviceCode).whenComplete((videoInfo, e)->{ realtime.realtime(deviceCode).whenComplete((videoInfo, e)->{
@ -416,7 +419,7 @@ public class Gb28181DownloadService {
CallIdHeader callId = provider.getNewCallId(); CallIdHeader callId = provider.getNewCallId();
String subscribeKey = GenericSubscribe.Helper.getKey(Request.INVITE, callId.getCallId()); String subscribeKey = GenericSubscribe.Helper.getKey(Request.INVITE, callId.getCallId());
subscribe.getInviteSubscribe().addPublisher(subscribeKey); subscribe.getInviteSubscribe().addPublisher(subscribeKey);
Flow.Subscriber<SIPResponse> subscriber = inviteSubscriber(docking,device,subscribeKey,cacheKey, ssrc, streamId, result, 60, TimeUnit.SECONDS); Flow.Subscriber<SIPResponse> subscriber = inviteSubscriber(docking,device,subscribeKey,cacheKey, ssrc, streamId, result, 0, TimeUnit.SECONDS);
subscribe.getInviteSubscribe().addSubscribe(subscribeKey, subscriber); subscribe.getInviteSubscribe().addSubscribe(subscribeKey, subscriber);
RedisUtil.StringOps.set(cacheKey, callId.getCallId()); RedisUtil.StringOps.set(cacheKey, callId.getCallId());
return SipRequestBuilder.createInviteRequest(ip, port, docking, device.getGbDeviceChannelId(), description.toString(), SipUtil.generateViaTag(), SipUtil.generateFromTag(), null, ssrc, callId); return SipRequestBuilder.createInviteRequest(ip, port, docking, device.getGbDeviceChannelId(), description.toString(), SipUtil.generateViaTag(), SipUtil.generateFromTag(), null, ssrc, callId);
@ -575,7 +578,12 @@ public class Gb28181DownloadService {
schedule[0].cancel(true); schedule[0].cancel(true);
} }
}; };
if(time == 0){
schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, 60, unit);
} else {
schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit); schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit);
}
return subscriber; return subscriber;
} }
@ -608,11 +616,16 @@ public class Gb28181DownloadService {
@Override @Override
public void onComplete() { public void onComplete() {
subscribe.getByeSubscribe().delPublisher(key); subscribe.getByeSubscribe().delPublisher(key);
if(time > 0){
schedule[0].cancel(true); schedule[0].cancel(true);
}
zlmMediaService.closeRtpServer(new CloseRtpServer(streamId)); zlmMediaService.closeRtpServer(new CloseRtpServer(streamId));
} }
}; };
if(time > 0){
schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit); schedule[0] = scheduledExecutorService.schedule(subscriber::onComplete, time, unit);
}
return subscriber; return subscriber;
} }
} }