ZLMRunner.java
6.97 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.MediaConfig;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import java.util.*;
@Component
@Order(value=1)
public class ZLMRunner implements CommandLineRunner {
private final static Logger logger = LoggerFactory.getLogger(ZLMRunner.class);
private Map<String, Boolean> startGetMedia;
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@Autowired
private ZLMHttpHookSubscribe hookSubscribe;
@Autowired
private IStreamProxyService streamProxyService;
@Autowired
private EventPublisher publisher;
@Autowired
private IMediaServerService mediaServerService;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private MediaConfig mediaConfig;
@Autowired
private DynamicTask dynamicTask;
@Override
public void run(String... strings) throws Exception {
mediaServerService.clearMediaServerForOnline();
MediaServerItem defaultMediaServer = mediaServerService.getDefaultMediaServer();
if (defaultMediaServer == null) {
mediaServerService.addToDatabase(mediaConfig.getMediaSerItem());
}else {
MediaServerItem mediaSerItem = mediaConfig.getMediaSerItem();
mediaServerService.updateToDatabase(mediaSerItem);
}
mediaServerService.syncCatchFromDatabase();
// 订阅 zlm启动事件, 新的zlm也会从这里进入系统
hookSubscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_server_started,new JSONObject(),
(MediaServerItem mediaServerItem, JSONObject response)->{
ZLMServerConfig zlmServerConfig = JSONObject.toJavaObject(response, ZLMServerConfig.class);
if (zlmServerConfig !=null ) {
if (startGetMedia != null) {
startGetMedia.remove(zlmServerConfig.getGeneralMediaServerId());
}
mediaServerService.zlmServerOnline(zlmServerConfig);
}
});
// 订阅 zlm保活事件, 当zlm离线时做业务的处理
hookSubscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_server_keepalive,new JSONObject(),
(MediaServerItem mediaServerItem, JSONObject response)->{
String mediaServerId = response.getString("mediaServerId");
if (mediaServerId !=null ) {
mediaServerService.updateMediaServerKeepalive(mediaServerId, response.getJSONObject("data"));
}
});
// 获取zlm信息
logger.info("[zlm接入]等待默认zlm中...");
// 获取所有的zlm， 并开启主动连接
List<MediaServerItem> all = mediaServerService.getAllFromDatabase();
mediaServerService.updateVmServer(all);
if (all.size() == 0) {
all.add(mediaConfig.getMediaSerItem());
}
for (MediaServerItem mediaServerItem : all) {
if (startGetMedia == null) {
startGetMedia = new HashMap<>();
}
startGetMedia.put(mediaServerItem.getId(), true);
connectZlmServer(mediaServerItem);
}
String taskKey = "zlm-connect-timeout";
dynamicTask.startDelay(taskKey, ()->{
if (startGetMedia != null) {
Set<String> allZlmId = startGetMedia.keySet();
for (String id : allZlmId) {
logger.error("[ {} ]]主动连接失败，不再尝试连接", id);
}
startGetMedia = null;
}
// TODO 清理数据库中与redis不匹配的zlm
}, 60 * 1000 );
}
@Async
public void connectZlmServer(MediaServerItem mediaServerItem){
String connectZlmServerTaskKey = "connect-zlm-" + mediaServerItem.getId();
ZLMServerConfig zlmServerConfigFirst = getMediaServerConfig(mediaServerItem);
if (zlmServerConfigFirst != null) {
zlmServerConfigFirst.setIp(mediaServerItem.getIp());
zlmServerConfigFirst.setHttpPort(mediaServerItem.getHttpPort());
startGetMedia.remove(mediaServerItem.getId());
mediaServerService.zlmServerOnline(zlmServerConfigFirst);
}else {
logger.info("[ {} ]-[ {}:{} ]主动连接失败, 清理相关资源， 开始尝试重试连接",
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
publisher.zlmOfflineEventPublish(mediaServerItem.getId());
}
dynamicTask.startCron(connectZlmServerTaskKey, ()->{
ZLMServerConfig zlmServerConfig = getMediaServerConfig(mediaServerItem);
if (zlmServerConfig != null) {
dynamicTask.stop(connectZlmServerTaskKey);
zlmServerConfig.setIp(mediaServerItem.getIp());
zlmServerConfig.setHttpPort(mediaServerItem.getHttpPort());
startGetMedia.remove(mediaServerItem.getId());
mediaServerService.zlmServerOnline(zlmServerConfig);
}
}, 2000);
}
public ZLMServerConfig getMediaServerConfig(MediaServerItem mediaServerItem) {
if (startGetMedia == null) { return null;}
if (!mediaServerItem.isDefaultServer() && mediaServerService.getOne(mediaServerItem.getId()) == null) {
return null;
}
if ( startGetMedia.get(mediaServerItem.getId()) == null || !startGetMedia.get(mediaServerItem.getId())) {
return null;
}
JSONObject responseJson = zlmresTfulUtils.getMediaServerConfig(mediaServerItem);
ZLMServerConfig zlmServerConfig = null;
if (responseJson != null) {
JSONArray data = responseJson.getJSONArray("data");
if (data != null && data.size() > 0) {
zlmServerConfig = JSON.parseObject(JSON.toJSONString(data.get(0)), ZLMServerConfig.class);
}
} else {
logger.error("[ {} ]-[ {}:{} ]主动连接失败, 2s后重试",
mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort());
}
return zlmServerConfig;
}
}