648540858
2021-11-12 eca1e05aeed3c51cce36bbce80c71bc3cbcea87d
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
package com.genersoft.iot.vmp.storager.impl;
 
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.storager.dao.*;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
 
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
/**    
 * @description:视频设备数据存储-jdbc实现
 * @author: swwheihei
 * @date:   2020年5月6日 下午2:31:42
 */
@SuppressWarnings("rawtypes")
@Component
public class VideoManagerStoragerImpl implements IVideoManagerStorager {
 
    private Logger logger = LoggerFactory.getLogger(VideoManagerStoragerImpl.class);
 
    @Autowired
    DataSourceTransactionManager dataSourceTransactionManager;
 
    @Autowired
    TransactionDefinition transactionDefinition;
 
    @Autowired
    private DeviceMapper deviceMapper;
 
    @Autowired
    private DeviceChannelMapper deviceChannelMapper;
 
    @Autowired
    private DeviceMobilePositionMapper deviceMobilePositionMapper;
 
    @Autowired
    private ParentPlatformMapper platformMapper;
 
    @Autowired
    private IRedisCatchStorage redisCatchStorage;
 
    @Autowired
    private PlatformChannelMapper platformChannelMapper;
 
    @Autowired
    private StreamProxyMapper streamProxyMapper;
 
    @Autowired
    private StreamPushMapper streamPushMapper;
 
    @Autowired
    private GbStreamMapper gbStreamMapper;
 
    @Autowired
    private VideoStreamSessionManager streamSession;
 
    @Autowired
    private MediaServerMapper mediaServerMapper;
 
    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
 
    /**
     * 根据设备ID判断设备是否存在
     *
     * @param deviceId 设备ID
     * @return true:存在  false:不存在
     */
    @Override
    public boolean exists(String deviceId) {
        return deviceMapper.getDeviceByDeviceId(deviceId) != null;
    }
 
    /**
     * 视频设备创建
     *
     * @param device 设备对象
     * @return true:创建成功  false:创建失败
     */
    @Override
    public synchronized boolean create(Device device) {
        return deviceMapper.add(device) > 0;
    }
 
 
 
    /**
     * 视频设备更新
     *
     * @param device 设备对象
     * @return true:更新成功  false:更新失败
     */
    @Override
    public synchronized boolean updateDevice(Device device) {
        String now = this.format.format(System.currentTimeMillis());
        device.setUpdateTime(now);
        Device deviceByDeviceId = deviceMapper.getDeviceByDeviceId(device.getDeviceId());
        if (deviceByDeviceId == null) {
            device.setCreateTime(now);
            return deviceMapper.add(device) > 0;
        }else {
            return deviceMapper.update(device) > 0;
        }
 
    }
 
    @Override
    public synchronized void updateChannel(String deviceId, DeviceChannel channel) {
        String channelId = channel.getChannelId();
        channel.setDeviceId(deviceId);
        channel.setStreamId(streamSession.getStreamId(deviceId, channel.getChannelId()));
        String now = this.format.format(System.currentTimeMillis());
        channel.setUpdateTime(now);
        DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
        if (deviceChannel == null) {
            channel.setCreateTime(now);
            deviceChannelMapper.add(channel);
        }else {
            deviceChannelMapper.update(channel);
        }
    }
 
    @Override
    public void deviceChannelOnline(String deviceId, String channelId) {
        deviceChannelMapper.online(deviceId, channelId);
    }
 
    @Override
    public void deviceChannelOffline(String deviceId, String channelId) {
        deviceChannelMapper.offline(deviceId, channelId);
    }
 
    @Override
    public void startPlay(String deviceId, String channelId, String streamId) {
        deviceChannelMapper.startPlay(deviceId, channelId, streamId);
    }
 
    @Override
    public void stopPlay(String deviceId, String channelId) {
        deviceChannelMapper.stopPlay(deviceId, channelId);
    }
 
    /**
     * 获取设备
     *
     * @param deviceId 设备ID
     * @return Device 设备对象
     */
    @Override
    public Device queryVideoDevice(String deviceId) {
        return deviceMapper.getDeviceByDeviceId(deviceId);
    }
 
    @Override
    public PageInfo queryChannelsByDeviceId(String deviceId, String query, Boolean hasSubChannel, Boolean online, int page, int count) {
        // 获取到所有正在播放的流
        PageHelper.startPage(page, count);
        List<DeviceChannel> all = deviceChannelMapper.queryChannelsByDeviceId(deviceId, null, query, hasSubChannel, online);
        return new PageInfo<>(all);
    }
 
    @Override
    public List<DeviceChannel> queryChannelsByDeviceId(String deviceId) {
        return deviceChannelMapper.queryChannelsByDeviceId(deviceId, null,null, null, null);
    }
 
    @Override
    public PageInfo<DeviceChannel> querySubChannels(String deviceId, String parentChannelId, String query, Boolean hasSubChannel, String online, int page, int count) {
        PageHelper.startPage(page, count);
        List<DeviceChannel> all = deviceChannelMapper.queryChannelsByDeviceId(deviceId, parentChannelId, null, null, null);
        return new PageInfo<>(all);
    }
 
    @Override
    public DeviceChannel queryChannel(String deviceId, String channelId) {
        return deviceChannelMapper.queryChannel(deviceId, channelId);
    }
 
 
    @Override
    public int delChannel(String deviceId, String channelId) {
        return deviceChannelMapper.del(deviceId, channelId);
    }
 
    /**
     * 获取多个设备
     *
     * @param page 当前页数
     * @param count 每页数量
     * @return PageInfo<Device> 分页设备对象数组
     */
    @Override
    public PageInfo<Device> queryVideoDeviceList(int page, int count) {
        PageHelper.startPage(page, count);
        List<Device> all = deviceMapper.getDevices();
        return new PageInfo<>(all);
    }
 
    /**
     * 获取多个设备
     *
     * @return List<Device> 设备对象数组
     */
    @Override
    public List<Device> queryVideoDeviceList() {
 
        List<Device> deviceList =  deviceMapper.getDevices();
        return deviceList;
    }
 
    /**
     * 删除设备
     *
     * @param deviceId 设备ID
     * @return true:删除成功  false:删除失败
     */
    @Override
    public boolean delete(String deviceId) {
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        boolean result = false;
        try {
            if (platformChannelMapper.delChannelForDeviceId(deviceId) <0  // 删除与国标平台的关联
                    || deviceChannelMapper.cleanChannelsByDeviceId(deviceId) < 0 // 删除他的通道
                    || deviceMapper.del(deviceId) < 0 // 移除设备信息
            ) {
                //事务回滚
                dataSourceTransactionManager.rollback(transactionStatus);
            }
            result = true;
            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
        }catch (Exception e) {
            dataSourceTransactionManager.rollback(transactionStatus);
        }
        return result;
    }
 
    /**
     * 更新设备在线
     *
     * @param deviceId 设备ID
     * @return true:更新成功  false:更新失败
     */
    @Override
    public synchronized boolean online(String deviceId) {
        Device device = deviceMapper.getDeviceByDeviceId(deviceId);
        if (device == null) {
            return false;
        }
        device.setOnline(1);
        logger.info("更新设备在线: " + deviceId);
        return deviceMapper.update(device) > 0;
    }
 
    /**
     * 更新设备离线
     *
     * @param deviceId 设备ID
     * @return true:更新成功  false:更新失败
     */
    @Override
    public synchronized boolean outline(String deviceId) {
        logger.info("更新设备离线: " + deviceId);
        Device device = deviceMapper.getDeviceByDeviceId(deviceId);
        if (device == null) return false;
        device.setOnline(0);
        return deviceMapper.update(device) > 0;
    }
 
    /**
     * 更新所有设备离线
     *
     * @return true:更新成功  false:更新失败
     */
    @Override
    public synchronized boolean outlineForAll() {
        logger.info("更新所有设备离线");
        int result = deviceMapper.outlineForAll();
        return result > 0;
    }
 
    /**
     * 清空通道
     * @param deviceId
     */
    @Override
    public void cleanChannelsForDevice(String deviceId) {
        deviceChannelMapper.cleanChannelsByDeviceId(deviceId);
    }
 
    /**
     * 添加Mobile Position设备移动位置
     * @param mobilePosition
     */
    @Override
    public synchronized boolean insertMobilePosition(MobilePosition mobilePosition) {
        return deviceMobilePositionMapper.insertNewPosition(mobilePosition) > 0;
    }
 
    /**
     * 查询移动位置轨迹
     * @param deviceId
     * @param startTime
     * @param endTime
     */
    @Override
    public synchronized List<MobilePosition> queryMobilePositions(String deviceId, String startTime, String endTime) {
        return deviceMobilePositionMapper.queryPositionByDeviceIdAndTime(deviceId, startTime, endTime);
    }
 
    @Override
    public boolean addParentPlatform(ParentPlatform parentPlatform) {
        int result = platformMapper.addParentPlatform(parentPlatform);
        return result > 0;
    }
 
    @Override
    public boolean updateParentPlatform(ParentPlatform parentPlatform) {
        int result = 0;
        ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId()); // .getDeviceGBId());
        if (parentPlatform.getId() == null ) {
            result = platformMapper.addParentPlatform(parentPlatform);
            if (parentPlatformCatch == null) {
                parentPlatformCatch = new ParentPlatformCatch();
                parentPlatformCatch.setParentPlatform(parentPlatform);
                parentPlatformCatch.setId(parentPlatform.getServerGBId());
            }
        }else {
            if (parentPlatformCatch == null) { // serverGBId 已变化
                ParentPlatform parentPlatById = platformMapper.getParentPlatById(parentPlatform.getId());
                // 使用旧的查出缓存ID
                parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatById.getServerGBId());
                parentPlatformCatch.setId(parentPlatform.getServerGBId());
                redisCatchStorage.delPlatformCatchInfo(parentPlatById.getServerGBId());
            }
            result = platformMapper.updateParentPlatform(parentPlatform);
        }
        // 更新缓存
        parentPlatformCatch.setParentPlatform(parentPlatform);
        redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
        return result > 0;
    }
 
    @Transactional
    @Override
    public boolean deleteParentPlatform(ParentPlatform parentPlatform) {
        int result = platformMapper.delParentPlatform(parentPlatform);
        // 删除关联的通道
        platformChannelMapper.cleanChannelForGB(parentPlatform.getServerGBId());
        return result > 0;
    }
 
    @Override
    public PageInfo<ParentPlatform> queryParentPlatformList(int page, int count) {
        PageHelper.startPage(page, count);
        List<ParentPlatform> all = platformMapper.getParentPlatformList();
        return new PageInfo<>(all);
    }
 
    @Override
    public ParentPlatform queryParentPlatByServerGBId(String platformGbId) {
        return platformMapper.getParentPlatByServerGBId(platformGbId);
    }
 
    @Override
    public List<ParentPlatform> queryEnableParentPlatformList(boolean enable) {
        return platformMapper.getEnableParentPlatformList(enable);
    }
 
    @Override
    public void outlineForAllParentPlatform() {
        platformMapper.outlineForAllParentPlatform();
    }
 
 
    @Override
    public PageInfo<ChannelReduce> queryAllChannelList(int page, int count, String query, Boolean online,
                                                       Boolean channelType, String platformId, Boolean inPlatform) {
        PageHelper.startPage(page, count);
        List<ChannelReduce> all = deviceChannelMapper.queryChannelListInAll(query, online, channelType, platformId, inPlatform);
        return new PageInfo<>(all);
    }
 
    @Override
    public List<ChannelReduce> queryChannelListInParentPlatform(String platformId) {
 
        return deviceChannelMapper.queryChannelListInAll(null, null, null, platformId, true);
    }
 
    @Override
    public int updateChannelForGB(String platformId, List<ChannelReduce> channelReduces) {
 
        Map<String, ChannelReduce> deviceAndChannels = new HashMap<>();
        for (ChannelReduce channelReduce : channelReduces) {
            deviceAndChannels.put(channelReduce.getDeviceId() + "_" + channelReduce.getChannelId(), channelReduce);
        }
        List<String> deviceAndChannelList = new ArrayList<>(deviceAndChannels.keySet());
        // 查询当前已经存在的
        List<String> relatedPlatformchannels = platformChannelMapper.findChannelRelatedPlatform(platformId, deviceAndChannelList);
        if (relatedPlatformchannels != null) {
            deviceAndChannelList.removeAll(relatedPlatformchannels);
        }
        for (String relatedPlatformchannel : relatedPlatformchannels) {
            deviceAndChannels.remove(relatedPlatformchannel);
        }
        List<ChannelReduce> channelReducesToAdd = new ArrayList<>(deviceAndChannels.values());
        // 对剩下的数据进行存储
        int result = 0;
        if (channelReducesToAdd.size() > 0) {
            result = platformChannelMapper.addChannels(platformId, channelReducesToAdd);
        }
 
        return result;
    }
 
 
    @Override
    public int delChannelForGB(String platformId, List<ChannelReduce> channelReduces) {
 
        int result = platformChannelMapper.delChannelForGB(platformId, channelReduces);
 
        return result;
    }
 
    @Override
    public DeviceChannel queryChannelInParentPlatform(String platformId, String channelId) {
        DeviceChannel channel = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
        return channel;
    }
 
    @Override
    public Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId) {
        Device device = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
        return device;
    }
 
    /**
     * 查询最新移动位置
     * @param deviceId
     */
    @Override
    public MobilePosition queryLatestPosition(String deviceId) {
        return deviceMobilePositionMapper.queryLatestPositionByDevice(deviceId);
    }
 
    /**
     * 删除指定设备的所有移动位置
     * @param deviceId
     */
    public int clearMobilePositionsByDeviceId(String deviceId) {
        return deviceMobilePositionMapper.clearMobilePositionsByDeviceId(deviceId);
    }
 
    /**
     * 新增代理流
     * @param streamProxyItem
     * @return
     */
    @Override
    public boolean addStreamProxy(StreamProxyItem streamProxyItem) {
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        boolean result = false;
        streamProxyItem.setStreamType("proxy");
        streamProxyItem.setStatus(true);
        String now = this.format.format(System.currentTimeMillis());
        streamProxyItem.setCreateTime(now);
        try {
            if (gbStreamMapper.add(streamProxyItem)<0 || streamProxyMapper.add(streamProxyItem) < 0) {
                //事务回滚
                dataSourceTransactionManager.rollback(transactionStatus);
            }
            result = true;
            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
        }catch (Exception e) {
            logger.error("向数据库添加流代理失败:", e);
            dataSourceTransactionManager.rollback(transactionStatus);
        }
        return result;
    }
 
    /**
     * 更新代理流
     * @param streamProxyItem
     * @return
     */
    @Override
    public boolean updateStreamProxy(StreamProxyItem streamProxyItem) {
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        boolean result = false;
        streamProxyItem.setStreamType("proxy");
        try {
            if (gbStreamMapper.update(streamProxyItem)<0 || streamProxyMapper.update(streamProxyItem) < 0) {
                //事务回滚
                dataSourceTransactionManager.rollback(transactionStatus);
            }
            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
            result = true;
        }catch (Exception e) {
            e.printStackTrace();
            dataSourceTransactionManager.rollback(transactionStatus);
        }
        return result;
    }
 
    /**
     * 移除代理流
     * @param app
     * @param stream
     * @return
     */
    @Override
    public int deleteStreamProxy(String app, String stream) {
        return streamProxyMapper.del(app, stream);
    }
 
    /**
     * 根据是否启用获取代理流列表
     * @param enable
     * @return
     */
    @Override
    public List<StreamProxyItem> getStreamProxyListForEnable(boolean enable) {
        return streamProxyMapper.selectForEnable(enable);
    }
 
    /**
     * 分页查询代理流列表
     * @param page
     * @param count
     * @return
     */
    @Override
    public PageInfo<StreamProxyItem> queryStreamProxyList(Integer page, Integer count) {
        PageHelper.startPage(page, count);
        List<StreamProxyItem> all = streamProxyMapper.selectAll();
        return new PageInfo<>(all);
    }
 
    /**
     * 根据国标ID获取平台关联的直播流
     * @param platformId
     * @param gbId
     * @return
     */
    @Override
    public GbStream queryStreamInParentPlatform(String platformId, String gbId) {
        return gbStreamMapper.queryStreamInPlatform(platformId, gbId);
    }
 
    /**
     * 获取平台关联的直播流
     * @param platformId
     * @return
     */
    @Override
    public List<GbStream> queryGbStreamListInPlatform(String platformId) {
        return gbStreamMapper.queryGbStreamListInPlatform(platformId);
    }
 
    /**
     * 按照是app和stream获取代理流
     * @param app
     * @param stream
     * @return
     */
    @Override
    public StreamProxyItem queryStreamProxy(String app, String stream){
        return streamProxyMapper.selectOne(app, stream);
    }
 
    @Override
    public void updateMediaList(List<StreamPushItem> streamPushItems) {
        if (streamPushItems == null || streamPushItems.size() == 0) return;
        logger.info("updateMediaList:  " + streamPushItems.size());
        streamPushMapper.addAll(streamPushItems);
        // TODO 待优化
        for (int i = 0; i < streamPushItems.size(); i++) {
            gbStreamMapper.setStatus(streamPushItems.get(i).getApp(), streamPushItems.get(i).getStream(), true);
        }
    }
 
    @Override
    public void updateMedia(StreamPushItem streamPushItem) {
        streamPushMapper.del(streamPushItem.getApp(), streamPushItem.getStream());
        streamPushMapper.add(streamPushItem);
        gbStreamMapper.setStatus(streamPushItem.getApp(), streamPushItem.getStream(), true);
    }
 
    @Override
    public void removeMedia(String app, String stream) {
        streamPushMapper.del(app, stream);
    }
 
    @Override
    public void clearMediaList() {
        streamPushMapper.clear();
    }
 
    @Override
    public void mediaOutline(String app, String streamId) {
        gbStreamMapper.setStatus(app, streamId, false);
    }
 
    @Override
    public void updateParentPlatformStatus(String platformGbID, boolean online) {
        platformMapper.updateParentPlatformStatus(platformGbID, online);
    }
 
    @Override
    public void updateMediaServer(MediaServerItem mediaServerItem) {
        String now = this.format.format(System.currentTimeMillis());
        mediaServerItem.setUpdateTime(now);
        if (mediaServerMapper.queryOne(mediaServerItem.getId()) != null) {
            mediaServerMapper.update(mediaServerItem);
        }else {
            mediaServerItem.setCreateTime(now);
            mediaServerMapper.add(mediaServerItem);
        }
    }
 
    @Override
    public List<StreamProxyItem> getStreamProxyListForEnableInMediaServer(String id, boolean enable) {
        return streamProxyMapper.selectForEnableInMediaServer(id, enable);
    }
 
 
    @Override
    public Device queryVideoDeviceByChannelId( String channelId) {
        Device result = null;
        List<DeviceChannel> channelList = deviceChannelMapper.queryChannelByChannelId(channelId);
        if (channelList.size() == 1) {
            result = deviceMapper.getDeviceByDeviceId(channelList.get(0).getDeviceId());
        }
        return result;
    }
}