sunpeng
/
stream_srs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: sunpeng:dashujuprod
Branches Tags
sunpeng:main
sunpeng:dashujuprod
sunpeng:miniotest
sunpeng:consulcheck
..
compare: sunpeng:main
Branches Tags
sunpeng:dashujuprod
sunpeng:main
sunpeng:miniotest
sunpeng:consulcheck

No commits in common. "dashujuprod" and "main" have entirely different histories.
dashujupro ... main

11 changed files with 724 additions and 702 deletions
Show Stats Expand all files Collapse all files

292
src/main/java/com/tuoheng/steam/controller/StreamRecordController.java
View File

@ -1,10 +1,6 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.controller;
import com.alibaba.fastjson2.JSON;
import com.tuoheng.steam.controller.dto.PageInfo;
import com.tuoheng.steam.controller.dto.PageStreamRequest;
@ -16,188 +12,162 @@ import com.tuoheng.steam.service.ITaskService;
import com.tuoheng.steam.service.dos.DayRecord;
import com.tuoheng.steam.service.dos.Mp4Record;
import com.tuoheng.steam.service.dos.StreamRecord;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.logging.log4j.util.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping({"/record"})
import java.io.File;
import java.util.*;
import java.util.stream.Collectors;
@RestController()
@RequestMapping("/record")
public class StreamRecordController {
private static final Logger logger = LoggerFactory.getLogger(StreamRecordController.class);
@Autowired
ITaskService taskService;
@Autowired
IRecordService iRecordService;
public StreamRecordController() {
}
@GetMapping({"pic"})
@GetMapping("pic")
public Response<String> startPic(@RequestParam String streamUrl) {
logger.info("启动视频拍照 :" + streamUrl);
streamUrl = this.streamSwitch(streamUrl);
logger.info("启动视频拍照_ :" + streamUrl);
if (Objects.isNull(streamUrl)) {
logger.info("启动视频拍照 :"+streamUrl);
streamUrl = streamSwitch(streamUrl);
logger.info("启动视频拍照_ :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
}
String outfile = taskService.startPic(streamUrl);
if(Objects.nonNull(outfile)) {
int lastSlashIndex = outfile.lastIndexOf("/");
String fileName = outfile.substring(lastSlashIndex + 1);
return Response.success(fileName);
}else {
return Response.fail(-1);
} else {
String outfile = this.taskService.startPic(streamUrl);
if (Objects.nonNull(outfile)) {
int lastSlashIndex = outfile.lastIndexOf("/");
String fileName = outfile.substring(lastSlashIndex + 1);
return Response.success(fileName);
} else {
return Response.fail(-1);
}
}
}
@GetMapping({"start"})
@GetMapping("start")
public Response<StreamTask> startRecording(@RequestParam String streamUrl) {
logger.info("启动录制 :" + streamUrl);
streamUrl = this.streamSwitch(streamUrl);
logger.info("启动录制_ :" + streamUrl);
if (Objects.isNull(streamUrl)) {
logger.info("启动录制 :"+streamUrl);
streamUrl = streamSwitch(streamUrl);
logger.info("启动录制_ :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
} else {
Response<StreamTask> response = Response.success(this.taskService.startTask(streamUrl));
logger.info("启动录制返回 :" + JSON.toJSONString(response));
return response;
}
Response<StreamTask> response = Response.success(taskService.startTask(streamUrl));
logger.info("启动录制返回 :"+ JSON.toJSONString(response));
return response;
}
@GetMapping({"stop"})
public Response<StreamTask> stopRecording(@RequestParam String streamUrl) {
logger.info("关闭录制 :" + streamUrl);
streamUrl = this.streamSwitch(streamUrl);
logger.info("关闭录制_ :" + streamUrl);
if (Objects.isNull(streamUrl)) {
return Response.fail(-1);
} else {
try {
Response<StreamTask> response = Response.success(this.taskService.stopTask(streamUrl));
if (Objects.isNull(response.getData()) || Objects.isNull(((StreamTask)response.getData()).getOutFileName()) || ((StreamTask)response.getData()).getOutFileName().isEmpty()) {
response.setCode(500);
}
@GetMapping("stop")
public Response<StreamTask> stopRecording(@RequestParam String streamUrl){
logger.info("关闭录制返回 :" + JSON.toJSONString(response));
return response;
} catch (Exception var3) {
return Response.fail(-1);
logger.info("关闭录制 :"+streamUrl);
streamUrl = streamSwitch(streamUrl);
logger.info("关闭录制_ :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
}
try {
Response<StreamTask> response = Response.success(taskService.stopTask(streamUrl));
if(Objects.isNull(response.getData()) || Objects.isNull(response.getData().getOutFileName()) || response.getData().getOutFileName().isEmpty()){
response.setCode(500);
}
}
}
@GetMapping({"info"})
public Response<StreamTask> getLastTask(@RequestParam String streamUrl) {
logger.info("查看录制 :" + streamUrl);
streamUrl = this.streamSwitch(streamUrl);
logger.info("查看录制_ :" + streamUrl);
if (Objects.isNull(streamUrl)) {
logger.info("关闭录制返回 :"+ JSON.toJSONString(response));
return response;
}catch (Exception e){
return Response.fail(-1);
} else {
Response<StreamTask> response = Response.success(this.taskService.getLastTask(streamUrl));
logger.info("查看录制返回 :" + JSON.toJSONString(response));
return response;
}
}
@GetMapping({"infoList"})
public Response<List<StreamTask>> infoList(@RequestParam String streamUrl) {
logger.info("查看录制列表 :" + streamUrl);
streamUrl = this.streamSwitch(streamUrl);
logger.info("查看录制列表_ :" + streamUrl);
if (Objects.isNull(streamUrl)) {
@GetMapping("info")
public Response<StreamTask> getLastTask(@RequestParam String streamUrl){
logger.info("查看录制 :"+streamUrl);
streamUrl = streamSwitch(streamUrl);
logger.info("查看录制_ :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
} else {
Response<List<StreamTask>> response = Response.success(this.taskService.getTaskList(streamUrl));
logger.info("查看录制列表返回 :" + JSON.toJSONString(response));
return response;
}
Response<StreamTask> response = Response.success(taskService.getLastTask(streamUrl));
logger.info("查看录制返回 :"+ JSON.toJSONString(response));
return response;
}
@PostMapping({"search"})
public Response<PageInfo<Mp4Info>> streamView(@RequestBody PageStreamRequest request) {
logger.info("查看录制 search :" + JSON.toJSONString(request));
if (!Objects.isNull(request.getPageIndex()) && !Objects.isNull(request.getPageSize()) && request.getPageIndex() >= 0 && request.getPageSize() > 0) {
List<Mp4Info> mp4s = this.searchAll(request.getStreamId());
if (!Strings.isBlank(request.getStartTime())) {
long filerTime = Long.parseLong(request.getStartTime()) - 900000L;
mp4s = (List)mp4s.stream().filter((info) -> info.getStartTime().compareTo(Long.toString(filerTime)) >= 0).collect(Collectors.toList());
}
if (!Strings.isBlank(request.getEndTime())) {
long filerTime = Long.parseLong(request.getEndTime());
mp4s = (List)mp4s.stream().filter((info) -> info.getStartTime().compareTo(Long.toString(filerTime)) <= 0).collect(Collectors.toList());
}
mp4s.sort((o1, o2) -> o2.getStartTime().compareTo(o1.getStartTime()));
PageInfo<Mp4Info> pageInfo = new PageInfo(mp4s, request.getPageIndex(), request.getPageSize());
logger.info("查看录制 search 返回:" + JSON.toJSONString(request));
return Response.success(pageInfo);
} else {
/**
* 废弃
* @param request
* @return
*/
@PostMapping("search")
public Response<PageInfo<Mp4Info>> streamView(@RequestBody PageStreamRequest request){
logger.info("查看录制 search :"+JSON.toJSONString(request));
if(Objects.isNull(request.getPageIndex()) || Objects.isNull(request.getPageSize())
|| request.getPageIndex() <0 || request.getPageSize() <=0 ){
logger.error("查看录制 search 入参错误！");
return Response.fail(-100);
}
List<Mp4Info> mp4s = searchAll(request.getStreamId());
if(!Strings.isBlank(request.getStartTime())){
long filerTime = Long.parseLong(request.getStartTime()) - 15 * 60 * 1000;
mp4s = mp4s.stream()
.filter(info -> info.getStartTime().compareTo(Long.toString(filerTime)) >= 0)
.collect(Collectors.toList()); // 将结果收集到列表中
}
if(!Strings.isBlank(request.getEndTime())){
long filerTime = Long.parseLong(request.getEndTime());
mp4s = mp4s.stream()
.filter(info -> info.getStartTime().compareTo(Long.toString(filerTime)) <= 0)
.collect(Collectors.toList()); // 将结果收集到列表中
}
/**
* 倒序排列
*/
mp4s.sort((o1, o2) -> o2.getStartTime().compareTo(o1.getStartTime()));
PageInfo<Mp4Info> pageInfo = new PageInfo<>(mp4s,request.getPageIndex(),request.getPageSize());
logger.info("查看录制 search 返回:"+JSON.toJSONString(request));
return Response.success(pageInfo);
}
@GetMapping({"history"})
public Response<List<Mp4Info>> streamView(@RequestParam String streamUrl) {
logger.info("查看录像历史返回 :" + streamUrl);
List<Mp4Info> dayMp4 = this.searchAll(streamUrl);
dayMp4.sort((o1, o2) -> o2.getStartTime().compareTo(o1.getStartTime()));
logger.info("查看录像历史返回_ :" + JSON.toJSONString(dayMp4));
return Response.success(dayMp4);
}
public List<Mp4Info> searchAll(String streamUrl) {
List<Mp4Info> dayMp4 = new ArrayList();
public List<Mp4Info> searchAll(String streamUrl){
for(DayRecord dayRecord : this.iRecordService.findDaysPath()) {
for(StreamRecord stream : this.iRecordService.findInDayRecord(dayRecord)) {
if (stream.getStreamId().equals(streamUrl)) {
List<Mp4Record> mp4Records = stream.queryMp4Records();
if (Objects.nonNull(mp4Records) && !mp4Records.isEmpty()) {
for(Mp4Record mp4Record : mp4Records) {
List<Mp4Info> dayMp4 = new ArrayList<>();
List<DayRecord> dayRecords = iRecordService.findDaysPath();
for(DayRecord dayRecord : dayRecords){
List<StreamRecord> streamRecords = iRecordService.findInDayRecord(dayRecord);
for(StreamRecord stream : streamRecords){
if(stream.getStreamId().equals(streamUrl)){
List<Mp4Record> mp4Records = stream.queryMp4Records();
if(Objects.nonNull(mp4Records) && !mp4Records.isEmpty()){
for(Mp4Record mp4Record : mp4Records){
Mp4Info mp4Info = new Mp4Info();
mp4Info.setStartTime(mp4Record.getStartTime());
String var10001 = dayRecord.getDay();
mp4Info.setUrl(var10001 + File.separator + stream.getStreamId() + File.separator + mp4Record.getMp4());
mp4Info.setUrl(dayRecord.getDay()+
File.separator + stream.getStreamId() + File.separator + mp4Record.getMp4());
dayMp4.add(mp4Info);
}
}
}
}
}
return dayMp4;
return dayMp4;
}
@GetMapping({"exit"})
public String exit() {
new Thread(() -> {
try {
this.taskService.stopAllTask();
Thread.sleep(10000L);
} catch (Exception var2) {
}
System.exit(0);
});
return "OK";
}
@Value("${srs.domain}")
private String srsdomain;
@ -205,60 +175,35 @@ public class StreamRecordController {
@Value("${srs.name}")
private String srsname;
// public String streamSwitch(String source) {
// if (source.contains("stream.t-aaron.com")) {
// return source;
// } else if (source.contains("rtmp://live.push.t-aaron.com")) {
// source = source.replace("rtmp://live.push.t-aaron.com", "http://live.play.t-aaron.com");
// if (source.endsWith("_")) {
// String var10000 = source.substring(0, source.length() - 1);
// source = var10000 + ".flv_";
// } else {
// source = source + ".flv";
// }
//
// return source;
// } else {
// if (source.contains("https://live.play.t-aaron.com")) {
// source = source.replace("https://live.play.t-aaron.com", "http://live.play.t-aaron.com");
// }
//
// return source;
// }
// }
public String streamSwitch(String source){
if(Objects.nonNull(srsname) && !srsname.isEmpty()){
if(Objects.nonNull(srsdomain) && !srsdomain.isEmpty()){
if(Objects.nonNull(source) && !source.isEmpty()){
return dockerFix(source);
return dockerFix(source);
}
}
if (source.contains("stream.t-aaron.com")) {
if(source.contains("stream.t-aaron.com")){
return source;
} else if (source.contains("rtmp://live.push.t-aaron.com")) {
source = source.replace("rtmp://live.push.t-aaron.com", "http://live.play.t-aaron.com");
if (source.endsWith("_")) {
String var10000 = source.substring(0, source.length() - 1);
source = var10000 + ".flv_";
} else {
} else if (source.contains("rtmp://live.push.t-aaron.com")){
source = source.replace("rtmp://live.push.t-aaron.com","http://live.play.t-aaron.com") ;
if(source.endsWith("_")){
source = source.substring(0,source.length()-1) + ".flv" + "_";
}else {
source = source + ".flv";
}
return source;
} else {
if (source.contains("https://live.play.t-aaron.com")) {
source = source.replace("https://live.play.t-aaron.com", "http://live.play.t-aaron.com");
}else {
if(source.contains("https://live.play.t-aaron.com")){
source = source.replace("https://live.play.t-aaron.com","http://live.play.t-aaron.com");
}
return source;
}
}
public String dockerFix(String url) {
logger.info("dockerFix{}",url);
if (url == null || url.isEmpty()) {
return url;
}
@ -278,7 +223,8 @@ public class StreamRecordController {
return withNewDomain.replaceFirst("\\.flv$", "");
}
return url;
}
}

6
src/main/java/com/tuoheng/steam/dos/StreamProcess.java
View File

@ -1,6 +1,5 @@
package com.tuoheng.steam.dos;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.tuoheng.steam.util.TimeUtils;
import java.io.Serializable;
@ -10,16 +9,13 @@ import java.util.concurrent.CompletableFuture;
public class StreamProcess implements Serializable {
@JsonIgnore
Process process;
Date createTime;
String stopTime;
ProcessType processType;
String fileName;
public Process getProcess() {
return process;
}
public String startTime(){
return TimeUtils.formatDateToString(createTime);

84
src/main/java/com/tuoheng/steam/dos/StreamTask.java
View File

@ -1,58 +1,54 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.dos;
import com.tuoheng.steam.util.TimeUtils;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class StreamTask implements Serializable {
Long taskId = generatePid();
Long taskId;
Date startTime;
String strStartTime;
String duration;
String stopCommandTime;
List<StreamProcess> streamProcesses = new ArrayList();
StreamProcess mergeProcess;
private static Long pid = 0L;
String streamUrl;
String outFileName;
String transaction;
public String getTransaction() {
return transaction;
}
public void setTransaction(String transaction) {
this.transaction = transaction;
}
public String getDuration() {
return this.duration;
return duration;
}
public void setDuration(String duration) {
this.duration = duration;
}
private static synchronized Long generatePid() {
if (pid == Long.MAX_VALUE) {
pid = 0L;
} else {
Long var0 = pid;
pid = pid + 1L;
}
String duration;
/**
* 接收到结束命令的时间
*/
String stopCommandTime;
/**
* 执行生成TS文件的命令
*/
List<StreamProcess> streamProcesses = new ArrayList<>();
/**
* 合并进程的命令
*/
StreamProcess mergeProcess;
private static Long pid = 0L;
private synchronized static Long generatePid(){
if(pid==Long.MAX_VALUE){
pid = 0L;
}else {
pid++;
}
return pid;
}
public String getStrStartTime() {
return this.strStartTime;
return strStartTime;
}
public void setStrStartTime(String strStartTime) {
@ -63,16 +59,19 @@ public class StreamTask implements Serializable {
this.strStartTime = startTime;
}
public String getStopCommandTime() {
return this.stopCommandTime;
return stopCommandTime;
}
public void setStopCommandTime(String stopCommandTime) {
this.stopCommandTime = stopCommandTime;
}
public List<StreamProcess> getStreamProcesses() {
return this.streamProcesses;
return streamProcesses;
}
public void setStreamProcesses(List<StreamProcess> streamProcesses) {
@ -80,7 +79,7 @@ public class StreamTask implements Serializable {
}
public StreamProcess getMergeProcess() {
return this.mergeProcess;
return mergeProcess;
}
public void setMergeProcess(StreamProcess mergeProcess) {
@ -88,15 +87,19 @@ public class StreamTask implements Serializable {
}
public String getStreamUrl() {
return this.streamUrl;
return streamUrl;
}
public void setStreamUrl(String streamUrl) {
this.streamUrl = streamUrl;
}
String streamUrl;
public Long getTaskId() {
return this.taskId;
return taskId;
}
public void setTaskId(Long taskId) {
@ -104,7 +107,7 @@ public class StreamTask implements Serializable {
}
public Date getStartTime() {
return this.startTime;
return startTime;
}
public void setStartTime(Date startTime) {
@ -112,17 +115,22 @@ public class StreamTask implements Serializable {
}
public StreamTask(String streamUrl) {
this.taskId = generatePid();
this.streamUrl = streamUrl;
this.startTime = new Date();
this.strStartTime = TimeUtils.formatDateToString(this.startTime);
this.stopCommandTime = null;
}
public String getOutFileName() {
return this.outFileName;
return outFileName;
}
public void setOutFileName(String outFileName) {
this.outFileName = outFileName;
}
String outFileName;
}

78
src/main/java/com/tuoheng/steam/schedule/Scheduler.java
View File

@ -1,62 +1,78 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.schedule;
import com.tuoheng.steam.service.IRecordService;
import com.tuoheng.steam.service.TaskService;
import com.tuoheng.steam.service.dos.DayRecord;
import com.tuoheng.steam.service.dos.FlvRecord;
import com.tuoheng.steam.service.dos.Mp4Record;
import com.tuoheng.steam.service.dos.StreamRecord;
import com.tuoheng.steam.util.TimeUtils;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.*;
@Component
public class Scheduler {
private static final Logger logger = LoggerFactory.getLogger(Scheduler.class);
@Autowired
IRecordService iRecordService;
@Value("${livedates}")
private Integer livedates;
@Value("${cangneiwai}")
private Boolean cangneiwai;
public Scheduler() {
}
@Scheduled(
fixedRate = 3600000L,
initialDelay = 6000L
)
/**
* 初次执行延迟6秒执行
* 每隔 60 分钟执行一次 60*60*1000
*/
@Scheduled(fixedRate = 3600000, initialDelay = 6000)
public void mergeTask() {
if (!this.cangneiwai) {
logger.info("舱内外无需录制");
} else {
Logger var10000 = logger;
long var10001 = System.currentTimeMillis();
var10000.info("开始FLV到MP4的转换 - " + var10001 / 1000L);
List<DayRecord> dayRecords = this.iRecordService.findDaysPath();
for(int index = 0; index < dayRecords.size(); ++index) {
DayRecord dayRecord = (DayRecord)dayRecords.get(index);
if (TimeUtils.isBefore(dayRecord.getDay(), this.livedates)) {
dayRecord.clear();
} else {
for(StreamRecord streamRecord : dayRecord.getStreamRecords()) {
for(FlvRecord flvRecord : streamRecord.queryFlvRecords()) {
this.iRecordService.mergeMp4(flvRecord);
}
if(!cangneiwai){
logger.info("舱内外无需录制");
return;
}
logger.info("开始FLV到MP4的转换 - " + System.currentTimeMillis() / 1000);
List<DayRecord> dayRecords = iRecordService.findDaysPath();
for (int index = 0; index < dayRecords.size(); index++) {
DayRecord dayRecord = dayRecords.get(index);
if(TimeUtils.isBefore(dayRecord.getDay(),livedates)){
dayRecord.clear();
} else {
List<StreamRecord> streamRecords = dayRecord.getStreamRecords();
for(StreamRecord streamRecord : streamRecords){
List<FlvRecord> flvRecords = streamRecord.queryFlvRecords();
for(FlvRecord flvRecord : flvRecords){
iRecordService.mergeMp4(flvRecord);
}
}
}
}
}
// public static boolean isWithin15Minutes(String timestamp1, String timestamp2) {
// // 将字符串转换为 long 类型
// long time1 = Long.parseLong(timestamp1);
// long time2 = Long.parseLong(timestamp2);
// // 计算时间差的绝对值
// long diff = Math.abs(time2 - time1);
// // 15 分钟 = 15 * 60 * 1000 毫秒
// long fifteenMinutesInMillis = 15 * 60 * 1000;
// // 判断是否小于或等于 15 分钟
// return diff <= fifteenMinutesInMillis;
// }
}

4
src/main/java/com/tuoheng/steam/service/ITaskService.java
View File

@ -2,8 +2,6 @@ package com.tuoheng.steam.service;
import com.tuoheng.steam.dos.StreamTask;
import java.util.List;
/**
* 视频录制服务
*/
@ -17,7 +15,5 @@ public interface ITaskService {
public StreamTask getLastTask(String streamUrl);
public List<StreamTask> getTaskList(String streamUrl);
public void stopAllTask();
}

51
src/main/java/com/tuoheng/steam/service/RecordService.java
View File

@ -1,66 +1,67 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.service;
import com.tuoheng.steam.service.dos.DayRecord;
import com.tuoheng.steam.service.dos.FlvRecord;
import com.tuoheng.steam.service.dos.StreamRecord;
import com.tuoheng.steam.service.dos.*;
import com.tuoheng.steam.service.innerService.ProcessService;
import com.tuoheng.steam.util.FileUtil;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.File;
import java.util.*;
import java.util.stream.Collectors;
import static com.tuoheng.steam.util.FileUtil.readFiles;
@Component
public class RecordService implements IRecordService {
@Value("${recordPath}")
private String recordPath;
@Autowired
ProcessService processService;
private static final Logger logger = LoggerFactory.getLogger(IRecordService.class);
public RecordService() {
}
public static boolean isValidDate(String dateStr) {
// 正则表达式：4位年份 + 2位月份 + 2位日期
String regex = "^\\d{4}(0[1-9]|1[0-2])(0[1-9]|[12][0-9]|3[01])$";
return dateStr.matches(regex);
}
@Override
public List<DayRecord> findDaysPath() {
List<DayRecord> days = new ArrayList();
for(String fileName : FileUtil.readFiles(this.recordPath)) {
if (isValidDate(fileName)) {
List<DayRecord> days = new ArrayList<>();
List<String> fileNames = readFiles(recordPath);
for (String fileName : fileNames) {
if(isValidDate(fileName)){
DayRecord day = new DayRecord();
day.setRoot(this.recordPath);
day.setRoot(recordPath);
day.setDay(fileName);
days.add(day);
}
}
days.sort(Comparator.comparing(DayRecord::getDay));
return days;
}
@Override
public List<StreamRecord> findInDayRecord(DayRecord dayRecord) {
return dayRecord.getStreamRecords();
return dayRecord.getStreamRecords();
}
@Override
public void mergeMp4(FlvRecord flvRecord) {
try {
this.processService.mergeMp4(flvRecord);
} catch (Exception e) {
processService.mergeMp4(flvRecord);
}catch (Exception e){
logger.error(e.getMessage());
}
}
}

408
src/main/java/com/tuoheng/steam/service/TaskService.java
View File

@ -1,8 +1,3 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.service;
import com.alibaba.fastjson2.JSON;
@ -12,17 +7,6 @@ import com.tuoheng.steam.service.innerService.ProcessService;
import com.tuoheng.steam.util.FileUtil;
import com.tuoheng.steam.util.ProcessManager;
import com.tuoheng.steam.util.TimeUtils;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -32,306 +16,352 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.PostConstruct;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.*;
@Service
@EnableScheduling
public class TaskService implements ITaskService {
public class TaskService implements ITaskService{
@Value("${srs.targetPath}")
private String targetPath;
private String ffmpeg = "ffmpeg";
// @Value("${ffmpeg}")
private String ffmpeg ="ffmpeg";
ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
private static final Logger logger = LoggerFactory.getLogger(TaskService.class);
@Autowired
ProcessService processService;
ConcurrentHashMap<String, StreamTask> runningTasks = new ConcurrentHashMap();
ConcurrentHashMap<String, Deque<StreamTask>> historyTasks = new ConcurrentHashMap();
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
public TaskService() {
}
@Scheduled(
fixedRate = 60000L
)
/**
* 清除没有被关闭的进程
*/
@Scheduled(fixedRate = 60000)
public void cleaTask() {
for(Map.Entry<String, StreamTask> entry : this.runningTasks.entrySet()) {
StreamTask streamTask = (StreamTask)entry.getValue();
if (streamTask.getStartTime().getTime() < (new Date()).getTime() - 5400000L) {
for (ConcurrentHashMap.Entry<String, StreamTask> entry : runningTasks.entrySet()) {
StreamTask streamTask = entry.getValue();
if(streamTask.getStartTime().getTime() < new Date().getTime() - 90 * 60 * 1000 ) {
logger.info("清理废弃任务 {}", JSON.toJSONString(streamTask));
try {
StreamTask s = this.stopTask((String)entry.getKey());
this.scheduler.schedule(() -> {
String var10002 = this.targetPath;
File file = new File(var10002 + "/" + s.getOutFileName());
StreamTask s = stopTask(entry.getKey());
scheduler.schedule(() -> {
File file = new File(targetPath +"/"+ s.getOutFileName());
try {
logger.info("废弃文件删除成功 {}", file.delete());
} catch (Exception e) {
logger.info("废弃文件删除成功 {}",file.delete());
}catch (Exception e) {
logger.error(e.getMessage());
}
}, 60, TimeUnit.SECONDS);
}catch (Exception ignore) {}
}, 60L, TimeUnit.SECONDS);
} catch (Exception var5) {
}
}
}
}
private static final Logger logger = LoggerFactory.getLogger(TaskService.class);
@Autowired
ProcessService processService;
/**
* 执行中的任务
*/
ConcurrentHashMap<String, StreamTask> runningTasks = new ConcurrentHashMap<>();
/**
* 历史任务
*/
ConcurrentHashMap<String, Deque<StreamTask>> historyTasks = new ConcurrentHashMap<>();
/**
* 任务池
*/
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
@PostConstruct
public void init() {
this.scheduler.scheduleWithFixedDelay(new Runnable() {
/**
*
*/
scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
TaskService.this.runningTasks.forEach((key, value) -> {
runningTasks.forEach((key, value) -> {
Calendar calendar = Calendar.getInstance();
calendar.add(10, -2);
calendar.add(Calendar.HOUR, -2);
Date twoHoursAgo = calendar.getTime();
if (value.getStartTime().before(twoHoursAgo)) {
TaskService.logger.error("taskId {} 执行超时,手动关闭", key);
logger.error("taskId {} 执行超时,手动关闭", key);
try {
TaskService.this.stopTask(key);
} catch (Exception var6) {
}
stopTask(key);
}catch (Exception ignore) {}
}
});
for(Map.Entry<String, Deque<StreamTask>> entry : TaskService.this.historyTasks.entrySet()) {
for(StreamTask task : entry.getValue()) {
for(StreamProcess process : task.getStreamProcesses()) {
for (Map.Entry<String, Deque<StreamTask>> entry : historyTasks.entrySet()) {
for (StreamTask task : entry.getValue()) {
List<StreamProcess> streamProcesses = task.getStreamProcesses();
for (StreamProcess process : streamProcesses) {
process.destroy();
File file = new File(process.getFileName());
FileUtil.deleteFile(file);
}
}
while(((Deque)entry.getValue()).size() > 10) {
((Deque)entry.getValue()).pollFirst();
while (entry.getValue().size()>10){
entry.getValue().pollFirst();
}
}
}
}, 60L, 60L, TimeUnit.SECONDS);
}, 60, 60, TimeUnit.SECONDS);
}
public String startPic(String streamUrl) {
try {
StreamProcess streamProcess = this.processService.takePic(streamUrl);
StreamProcess streamProcess = processService.takePic(streamUrl);
String outFileName = streamProcess.getFileName();
/*
注册进程
*/
ProcessManager.registerProcess(streamProcess.getInnerProcessId());
logger.info("streamUrl {} startPicProcess {} ", streamUrl, streamProcess.getInnerProcessId());
logger.info("streamUrl {} startPicProcess {} ", streamUrl,streamProcess.getInnerProcessId());
CompletableFuture<Process> future = streamProcess.onExit();
future.thenRun(() -> {
logger.info("streamUrl {} startPicProcess {} Over ", streamUrl, streamProcess.getInnerProcessId());
logger.info("streamUrl {} startPicProcess {} Over ", streamUrl,streamProcess.getInnerProcessId());
streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
}).exceptionally((ex) -> {
logger.info("streamUrl {} startPicProcess {} Exceptionally ", streamUrl, streamProcess.getInnerProcessId());
}).exceptionally(ex -> {
logger.info("streamUrl {} startPicProcess {} Exceptionally ", streamUrl,streamProcess.getInnerProcessId());
streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
return null;
});
return outFileName;
} catch (Exception e) {
logger.info("streamUrl {} ", streamUrl, e);
return null;
}catch (Exception e) {
logger.info("streamUrl {} ", streamUrl,e);
}
return null;
}
public StreamTask startTask(String streamUrl) {
if (this.runningTasks.containsKey(streamUrl)) {
return (StreamTask)this.runningTasks.get(streamUrl);
} else {
StreamTask taskInstance = new StreamTask(streamUrl);
// taskInstance.setTransaction(transaction);
if (this.runningTasks.putIfAbsent(streamUrl, taskInstance) == null) {
logger.info("streamUrl {} startTask {} transaction {}", streamUrl, taskInstance.getTaskId(), taskInstance.getTransaction());
this.startTask(streamUrl, taskInstance);
return taskInstance;
} else {
return (StreamTask)this.runningTasks.get(streamUrl);
}
/*
当前有任务,则返回当前的任务
*/
if (runningTasks.containsKey(streamUrl)) {
return runningTasks.get(streamUrl);
}
/*
当前无任务,则开启新任务
*/
StreamTask taskInstance = new StreamTask(streamUrl);
if (runningTasks.putIfAbsent(streamUrl, taskInstance) == null) {
logger.info("streamUrl {} startTask {} ", streamUrl, taskInstance.getTaskId());
startTask(streamUrl, taskInstance);
// runningTasks.put(streamUrl, taskInstance);
return taskInstance;
} else {
return runningTasks.get(streamUrl);
}
}
public StreamTask stopTask(final String streamUrl) throws Exception {
boolean recordSuccess = true;
final StreamTask currentStreamTask = (StreamTask)this.runningTasks.remove(streamUrl);
if (currentStreamTask == null) {
throw new Exception("");
} else {
currentStreamTask.setStopCommandTime(TimeUtils.formatDateToString(new Date()));
if (!CollectionUtils.isEmpty(currentStreamTask.getStreamProcesses())) {
for(StreamProcess streamProcess : currentStreamTask.getStreamProcesses()) {
if (!(new File(streamProcess.getFileName())).exists()) {
recordSuccess = false;
logger.error("流录制失败: streamUrl {} taskId {} destroy Process {}", new Object[]{streamUrl, currentStreamTask.getTaskId(), streamProcess.getInnerProcessId()});
} else {
logger.info("streamUrl {} taskId {} destroy Process {}", new Object[]{streamUrl, currentStreamTask.getTaskId(), streamProcess.getInnerProcessId()});
}
public StreamTask stopTask(String streamUrl) throws Exception {
boolean recordSuccess = true;
StreamTask currentStreamTask = runningTasks.remove(streamUrl);
if (currentStreamTask!= null) {
currentStreamTask.setStopCommandTime(TimeUtils.formatDateToString(new Date()));
if (!CollectionUtils.isEmpty(currentStreamTask.getStreamProcesses())) {
for (StreamProcess streamProcess : currentStreamTask.getStreamProcesses()) {
if(!new File(streamProcess.getFileName()).exists()){
recordSuccess = false;
logger.error("流录制失败: streamUrl {} taskId {} destroy Process {}", streamUrl,currentStreamTask.getTaskId() ,streamProcess.getInnerProcessId());
}else {
logger.info("streamUrl {} taskId {} destroy Process {}", streamUrl,currentStreamTask.getTaskId() ,streamProcess.getInnerProcessId());
}
streamProcess.destroy();
}
}
if (!recordSuccess) {
if(!recordSuccess){
throw new Exception("");
} else {
try {
String outFileName = UUID.randomUUID().toString() + ".mp4";
StreamProcess mergeProcess = this.processService.mergeStream(outFileName, currentStreamTask.getStreamProcesses());
if (Objects.nonNull(mergeProcess)) {
ProcessManager.registerProcess(mergeProcess.getInnerProcessId());
currentStreamTask.setMergeProcess(mergeProcess);
logger.info("streamUrl {} taskId {} startMergeProcess {} ", new Object[]{streamUrl, currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId()});
CompletableFuture<Process> future = mergeProcess.onExit();
future.thenRun(() -> {
logger.info("streamUrl {} taskId {} MergeProcess {} Over", new Object[]{streamUrl, currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId()});
mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
}
for(StreamProcess streamProcess : currentStreamTask.getStreamProcesses()) {
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
}
try {
String outFileName = UUID.randomUUID().toString() + ".mp4";
StreamProcess mergeProcess = processService.mergeStream(outFileName, currentStreamTask.getStreamProcesses());
}).exceptionally((ex) -> {
logger.error("streamUrl {} taskId {} MergeProcess {} exceptionally", new Object[]{streamUrl, currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId(), ex});
mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
if(Objects.nonNull(mergeProcess)){
/**
* 注册进程
*/
ProcessManager.registerProcess(mergeProcess.getInnerProcessId());
currentStreamTask.setMergeProcess(mergeProcess);
for(StreamProcess streamProcess : currentStreamTask.getStreamProcesses()) {
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
}
return null;
});
currentStreamTask.setOutFileName(outFileName);
}
logger.info("streamUrl {} taskId {} startMergeProcess {} ", streamUrl, currentStreamTask.getTaskId(),
mergeProcess.getInnerProcessId());
if (Objects.nonNull(currentStreamTask.getMergeProcess()) && Objects.nonNull(currentStreamTask.getMergeProcess().getFileName()) && !currentStreamTask.getMergeProcess().getFileName().isEmpty()) {
CompletableFuture<Process> future = mergeProcess.onExit();
future.thenRun(() -> {
logger.info("streamUrl {} taskId {} MergeProcess {} Over", streamUrl,currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId());
mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
}
}).exceptionally(ex -> {
logger.error("streamUrl {} taskId {} MergeProcess {} exceptionally", streamUrl,currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId(), ex);
mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
}
return null;
});
currentStreamTask.setOutFileName(outFileName);
}
if(Objects.nonNull(currentStreamTask.getMergeProcess())){
if(Objects.nonNull(currentStreamTask.getMergeProcess().getFileName()) &&
!currentStreamTask.getMergeProcess().getFileName().isEmpty()){
try {
this.executor.schedule(new Runnable() {
executor.schedule(new Runnable() {
@Override
public void run() {
try {
TaskService.logger.info("FileName :" + currentStreamTask.getMergeProcess().getFileName());
Runtime var10000 = Runtime.getRuntime();
String var10001 = TaskService.this.ffmpeg;
Process process = var10000.exec(var10001 + " -i " + currentStreamTask.getMergeProcess().getFileName());
logger.info("FileName :"+currentStreamTask.getMergeProcess().getFileName());
Process process = Runtime.getRuntime().exec(ffmpeg + " -i " +
currentStreamTask.getMergeProcess().getFileName());
BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
String line;
while((line = reader.readLine()) != null) {
TaskService.logger.info("Duration line: " + line);
while ((line = reader.readLine()) != null) {
logger.info("Duration line: " + line);
if (line.contains("Duration")) {
String duration = line.split("Duration: ")[1].split(",")[0];
TaskService.logger.info("Duration: " + duration.substring(0, 8));
logger.info("Duration: " + duration.substring(0, 8));
currentStreamTask.setDuration(duration.substring(0, 8));
TaskService.logger.info("-----------------放入缓存-----------------");
if (TaskService.this.historyTasks.containsKey(streamUrl)) {
((Deque)TaskService.this.historyTasks.get(streamUrl)).add(currentStreamTask);
} else {
TaskService.this.historyTasks.put(streamUrl, new LinkedList());
((Deque)TaskService.this.historyTasks.get(streamUrl)).offerLast(currentStreamTask);
logger.info("-----------------放入缓存-----------------");
/**
* 放入缓存
*/
if( historyTasks.containsKey(streamUrl)){
historyTasks.get(streamUrl).add(currentStreamTask);
}else {
historyTasks.put(streamUrl, new LinkedList<>());
historyTasks.get(streamUrl).offerLast(currentStreamTask);
}
break;
}
}
reader.close();
} catch (Exception e) {
TaskService.logger.info(e.getMessage());
}catch (Exception e){
logger.info(e.getMessage());
}
}
}, 40L, TimeUnit.SECONDS);
},40, TimeUnit.SECONDS);
} catch (Exception e) {
e.printStackTrace();
}
}
return currentStreamTask;
} catch (Exception e) {
logger.info("taskId {} Stop exceptionally", streamUrl, e);
return null;
}
}
}
}
public List<StreamTask> getTaskList(String streamUrl){
List<StreamTask> ret = new ArrayList<>();
StreamTask current = (StreamTask)this.runningTasks.get(streamUrl);
if(Objects.nonNull(current)){
ret.add(current);
}
if(this.historyTasks.containsKey(streamUrl)) {
List<StreamTask> historys = new ArrayList<>(this.historyTasks.get(streamUrl));
if(!historys.isEmpty()) {
ret.addAll(historys);
return currentStreamTask;
} catch (Exception e) {
logger.info("taskId {} Stop exceptionally", streamUrl,e);
return null;
}
}else {
throw new Exception("");
// if(historyTasks.containsKey(streamUrl)){
// return historyTasks.get(streamUrl).peekLast();
// }else {
// return null;
// }
}
return ret;
}
public StreamTask getLastTask(String streamUrl) {
StreamTask current = (StreamTask)this.runningTasks.get(streamUrl);
if (Objects.nonNull(current)) {
StreamTask current = runningTasks.get(streamUrl);
if(Objects.nonNull(current)){
return current;
} else {
return this.historyTasks.containsKey(streamUrl) ? (StreamTask)((Deque)this.historyTasks.get(streamUrl)).peekLast() : null;
}else {
if(historyTasks.containsKey(streamUrl)){
return historyTasks.get(streamUrl).peekLast();
}
return null;
}
}
private void startTask(String streamUrl, StreamTask streamTask) {
try {
/*
任务不存在了,直接退出
*/
if (streamTask.getStopCommandTime() != null) {
logger.info("streamUrl {} Task {} Stopped", streamUrl, streamTask.getTaskId());
return;
}
StreamProcess streamProcess = this.processService.recordStream(streamUrl);
StreamProcess streamProcess = processService.recordStream(streamUrl);
/**
* 注册进程
*/
ProcessManager.registerProcess(streamProcess.getInnerProcessId());
streamTask.getStreamProcesses().add(streamProcess);
logger.info("streamUrl {} taskId {} startProcess {} ", new Object[]{streamUrl, streamTask.getTaskId(), streamProcess.getInnerProcessId()});
logger.info("streamUrl {} taskId {} startProcess {} ", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId());
CompletableFuture<Process> future = streamProcess.onExit();
future.thenRun(() -> {
logger.info("streamUrl {} taskId {} Process {} Over", new Object[]{streamUrl, streamTask.getTaskId(), streamProcess.getInnerProcessId()});
logger.info("streamUrl {} taskId {} Process {} Over", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId());
streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
try {
Thread.sleep(2000L);
} catch (Exception var5) {
}
if (!(new File(streamProcess.getFileName())).exists()) {
} catch (Exception ignored){}
if(!new File(streamProcess.getFileName()).exists()){
streamTask.getStreamProcesses().remove(streamProcess);
}
this.startTask(streamUrl, streamTask);
}).exceptionally((ex) -> {
logger.info("streamUrl {} taskId {} Process {} Exceptionally", new Object[]{streamUrl, streamTask.getTaskId(), streamProcess.getInnerProcessId(), ex});
startTask(streamUrl, streamTask);
}).exceptionally(ex -> {
logger.info("streamUrl {} taskId {} Process {} Exceptionally", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId(),ex);
streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
this.startTask(streamUrl, streamTask);
startTask(streamUrl, streamTask);
ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
return null;
});
} catch (IOException e) {
logger.info("streamUrl {} taskId {} startProcess Exception", new Object[]{streamUrl, streamTask.getTaskId(), e});
logger.info("streamUrl {} taskId {} startProcess Exception", streamUrl,streamTask.getTaskId(),e);
}
}
public void stopAllTask() {
this.runningTasks.forEach((key, value) -> {
try {
this.stopTask(key);
} catch (Exception var4) {
}
public void stopAllTask(){
runningTasks.forEach((key, value) -> {
try {
stopTask(key);
}catch (Exception ignore){}
});
}
}

287
src/main/java/com/tuoheng/steam/service/innerService/ProcessService.java
View File

@ -1,212 +1,249 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.service.innerService;
import com.tuoheng.steam.dos.ProcessType;
import com.tuoheng.steam.dos.StreamProcess;
import com.tuoheng.steam.dos.ProcessType;
import com.tuoheng.steam.service.dos.FlvRecord;
import com.tuoheng.steam.util.FileUtil;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import com.tuoheng.steam.util.ProcessManager;
import com.tuoheng.steam.util.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
@Service
public class ProcessService {
@Value("${srs.splitPath}")
private String splitPath;
// @Value("${ffmpeg}")
private String ffmpeg = "ffmpeg";
@Value("${srs.targetPath}")
private String targetPath;
@Value("${recordPath}")
private String recordPath;
@Value("${video.bitrate:2000k}")
private String videoBitrate;
@Value("${audio.bitrate:128k}")
private String audioBitrate;
private static final Logger logger = LoggerFactory.getLogger(ProcessService.class);
/**
* 任务池
*/
ExecutorService loggingService = Executors.newCachedThreadPool();
public ProcessService() {
}
public void mergeMp4(FlvRecord flvRecord) throws IOException, ExecutionException, InterruptedException {
String command = ffmpeg + " -i "+ recordPath+ File.separator +
flvRecord.getStream().getDayRecord().getDay() + File.separator +
flvRecord.getStream().getStreamId() + File.separator + flvRecord.getFlv() + " -c:v copy -c:a aac " + recordPath+ File.separator +
flvRecord.getStream().getDayRecord().getDay() + File.separator +
flvRecord.getStream().getStreamId() + File.separator +flvRecord.getStartTime() + ".mp4";
public void mergeMp4(final FlvRecord flvRecord) throws IOException, ExecutionException, InterruptedException {
String var10000 = this.ffmpeg;
final String command = var10000 + " -i " + this.recordPath + File.separator + flvRecord.getStream().getDayRecord().getDay() + File.separator + flvRecord.getStream().getStreamId() + File.separator + flvRecord.getFlv() + " -c:v copy -c:a aac " + this.recordPath + File.separator + flvRecord.getStream().getDayRecord().getDay() + File.separator + flvRecord.getStream().getStreamId() + File.separator + flvRecord.getStartTime() + ".mp4";
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
pb.redirectErrorStream(true);
final Process process = pb.start();
this.loggingService.execute(new Runnable() {
public void run() {
ProcessService.logger.info("mergeMp4 process Start {}", command);
Process process = pb.start();
String line;
try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
while((line = reader.readLine()) != null) {
ProcessService.logger.info("mergeMp4-------- {}", line);
loggingService.execute(new Runnable() {
@Override
public void run() {
logger.info("mergeMp4 process Start {}",command);
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(process.getInputStream()))) {
String line;
while ((line = reader.readLine()) != null) {
logger.info("mergeMp4-------- {}",line);
}
} catch (IOException e) {
ProcessService.logger.info("mergeMp4-------- Over", e);
logger.info("mergeMp4-------- Over",e);
}
ProcessService.logger.info("mergeMp4 Over");
String var10002 = ProcessService.this.recordPath;
File delete = new File(var10002 + File.separator + flvRecord.getStream().getDayRecord().getDay() + File.separator + flvRecord.getStream().getStreamId() + File.separator + flvRecord.getFlv());
logger.info("mergeMp4 Over");
File delete = new File(recordPath+ File.separator +
flvRecord.getStream().getDayRecord().getDay() + File.separator +
flvRecord.getStream().getStreamId() + File.separator + flvRecord.getFlv());
FileUtil.deleteFile(delete);
}
});
CompletableFuture<Process> future = process.onExit();
Process completedProcess = null;
CompletableFuture<Process> future = process.onExit();
// 阻塞等待进程结束
Process completedProcess = null;
try {
completedProcess = (Process)future.get(4L, TimeUnit.SECONDS);
logger.info("mergeMp4正常完成-------- Over");
} catch (Exception e) {
var10000 = this.recordPath;
String fileName = var10000 + File.separator + flvRecord.getStream().getDayRecord().getDay() + File.separator + flvRecord.getStream().getStreamId() + File.separator + flvRecord.getStartTime() + ".mp4";
completedProcess = future.get(4, TimeUnit.SECONDS);
logger.info("mergeMp4正常完成-------- Over");
}catch (Exception e) {
String fileName = recordPath+ File.separator +
flvRecord.getStream().getDayRecord().getDay() + File.separator +
flvRecord.getStream().getStreamId() + File.separator +flvRecord.getStartTime() + ".mp4";
File file = new File(fileName);
if (file.exists()) {
logger.info("mergeMp4超时4S完成-------- Over", e);
if(file.exists()){
logger.info("mergeMp4超时4S完成-------- Over",e);
process.destroy();
} else {
try {
completedProcess = (Process)future.get(4L, TimeUnit.SECONDS);
} catch (Exception var11) {
if (file.exists()) {
logger.info("mergeMp4超时8S完成-------- Over", e);
completedProcess = future.get(4, TimeUnit.SECONDS);
}catch (Exception e1) {
if(file.exists()){
logger.info("mergeMp4超时8S完成-------- Over",e);
process.destroy();
} else {
logger.info("mergeMp4超时8S未完成-------- Over", e);
}else {
logger.info("mergeMp4超时8S未完成-------- Over",e);
process.destroy();
}
}
}
}
if (Objects.nonNull(completedProcess)) {
if(Objects.nonNull(completedProcess)) {
// 检查进程是否成功结束
if (completedProcess.exitValue() == 0) {
logger.info("进程成功结束！");
} else {
logger.info("进程失败，退出码 {} ", completedProcess.exitValue());
logger.info("进程失败，退出码 {} " ,completedProcess.exitValue());
}
}
}
public StreamProcess takePic(String streamUrl) throws IOException {
String var10000 = this.targetPath;
String outFileName = var10000 + "/" + UUID.randomUUID().toString() + ".jpg";
String command = String.format(this.ffmpeg + " -i %s -vf fps=1 -frames:v 1 -q:v 2 %s", streamUrl, outFileName);
List<String> listCommand = new ArrayList();
listCommand.add(this.ffmpeg);
String outFileName = targetPath + "/" +UUID.randomUUID().toString() +".jpg";
String command = String.format(
ffmpeg+ " -i %s -vf fps=1 -frames:v 1 -q:v 2 %s",
streamUrl, outFileName);
List<String> listCommand = new ArrayList<>();
listCommand.add(ffmpeg); // ffmpeg 路径
listCommand.add("-i");
listCommand.add(streamUrl);
listCommand.add(streamUrl); // 流地址
listCommand.add("-vf");
listCommand.add("fps=1");
listCommand.add("-frames:v");
listCommand.add("1");
listCommand.add("-q:v");
listCommand.add("2");
listCommand.add(outFileName);
listCommand.add(outFileName); // 输出文件名
logger.info("takePic {}", listCommand);
ProcessBuilder pb = new ProcessBuilder(listCommand);
pb.redirectErrorStream(true);
final Process process = pb.start();
this.loggingService.execute(new Runnable() {
public void run() {
String line;
try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
while((line = reader.readLine()) != null) {
ProcessService.logger.info("recordStream-------- {}", line);
}
} catch (IOException var6) {
ProcessService.logger.info("takePic-------- Over");
}
Process process = pb.start();
loggingService.execute(new Runnable() {
@Override
public void run() {
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(process.getInputStream()))) {
String line;
while ((line = reader.readLine()) != null) {
logger.info("recordStream-------- {}",line);
}
} catch (IOException e) {
logger.info("takePic-------- Over");
}
}
});
return new StreamProcess(process, outFileName, ProcessType.PIC);
return new StreamProcess(process,outFileName, ProcessType.PIC);
}
public StreamProcess recordStream(String streamUrl) throws IOException {
String var10000 = this.splitPath;
String recordFileName = var10000 + "/" + UUID.randomUUID().toString() + ".ts";
String command = String.format(this.ffmpeg + " -i %s -c:v libx264 -b:v %s -c:a aac -b:a %s -f mpegts -reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 2 %s", Optional.of(streamUrl).filter((s) -> s.endsWith("_")).map((s) -> s.substring(0, s.length() - 1)).orElse(streamUrl), this.videoBitrate, this.audioBitrate, recordFileName);
String recordFileName = splitPath+'/'+UUID.randomUUID().toString() +".ts";
/**
* 如果streamUrl末尾有个_,在实际录制的时候需要将_去除
*/
String command = String.format(
ffmpeg+ " -i %s -c copy -f mpegts -reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 2 %s",
Optional.of(streamUrl)
.filter(s -> s.endsWith("_"))
.map(s -> s.substring(0, s.length() - 1))
.orElse(streamUrl), recordFileName);
logger.info("recordStream {}", command);
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
pb.redirectErrorStream(true);
final Process process = pb.start();
this.loggingService.execute(new Runnable() {
public void run() {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
while(reader.readLine() != null) {
}
} catch (IOException var6) {
ProcessService.logger.info("recordStream-------- Over");
}
Process process = pb.start();
loggingService.execute(new Runnable() {
@Override
public void run() {
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(process.getInputStream()))) {
String line;
while ((line = reader.readLine()) != null) {
// logger.info("recordStream-------- {}",line);
}
} catch (IOException e) {
logger.info("recordStream-------- Over");
}
}
});
return new StreamProcess(process, recordFileName, ProcessType.RECORD);
}
return new StreamProcess(process,recordFileName, ProcessType.RECORD);
};
public StreamProcess mergeStream(String outFileName, List<StreamProcess> streamProcesses) throws IOException {
List<String> fileList = new ArrayList();
for(StreamProcess streamProcess : streamProcesses) {
List<String> fileList = new ArrayList<>();
for(StreamProcess streamProcess : streamProcesses){
File newFile = new File(streamProcess.getFileName());
if (newFile.exists()) {
if(newFile.exists()){
fileList.add(streamProcess.getFileName());
}
}
if (fileList.isEmpty()) {
if(fileList.isEmpty()){
return null;
} else {
String command = "";
outFileName = this.targetPath + "/" + outFileName;
if (fileList.size() == 1) {
command = String.format(this.ffmpeg + " -i %s -c copy %s", fileList.get(0), outFileName);
} else {
String filePaths = String.join("|", fileList);
command = String.format(this.ffmpeg + " -i \"concat:%s\" -c copy %s", filePaths, outFileName);
}
logger.info("mergeStream {}", command);
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
pb.redirectErrorStream(true);
final Process process = pb.start();
this.loggingService.execute(new Runnable() {
public void run() {
String line;
try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
while((line = reader.readLine()) != null) {
ProcessService.logger.info("mergeStream-------- {}", line);
}
} catch (IOException var6) {
ProcessService.logger.info("mergeStream-------- Over");
}
}
});
return new StreamProcess(process, outFileName, ProcessType.MERGE);
}
}
String command = "";
outFileName = targetPath + "/" +outFileName;
if(fileList.size() == 1){
command = String.format(
ffmpeg+ " -i %s -c copy %s",
fileList.get(0), outFileName);
}else {
String filePaths = String.join("|", fileList);
command = String.format(
ffmpeg+ " -i \"concat:%s\" -c copy %s",
filePaths, outFileName);
}
logger.info("mergeStream {}", command);
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
pb.redirectErrorStream(true);
Process process = pb.start();
loggingService.execute(new Runnable() {
@Override
public void run() {
try (BufferedReader reader = new BufferedReader(
new InputStreamReader(process.getInputStream()))) {
String line;
while ((line = reader.readLine()) != null) {
logger.info("mergeStream-------- {}",line);
}
} catch (IOException e) {
logger.info("mergeStream-------- Over");
}
}
});
return new StreamProcess(process,outFileName,ProcessType.MERGE);
};
}

73
src/main/java/com/tuoheng/steam/util/FileUtil.java
View File

@ -1,95 +1,100 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.util;
import com.tuoheng.steam.controller.StreamRecordController;
import com.tuoheng.steam.service.dos.DayRecord;
import com.tuoheng.steam.service.dos.StreamType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.LinkOption;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class FileUtil {
private static final Logger logger = LoggerFactory.getLogger(FileUtil.class);
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
public FileUtil() {
}
private static final Logger logger = LoggerFactory.getLogger(FileUtil.class);
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
public static void deleteFile(File file) {
if (file.exists()) {
if(file.exists()) {
file.delete();
}
}
public static void deleteFolder(String stringPath) {
try {
Path path = Paths.get(stringPath);
if (Files.exists(path, new LinkOption[0])) {
if (Files.exists(path)) {
try (Stream<Path> pathStream = Files.walk(path)) {
pathStream.sorted((p1, p2) -> -p1.compareTo(p2)).forEach((p) -> {
try {
Files.delete(p);
} catch (IOException e) {
System.err.println("无法删除: " + p + ", 原因: " + e.getMessage());
}
});
pathStream.sorted((p1, p2) -> -p1.compareTo(p2)) // 从子文件/文件夹开始删除
.forEach(p -> {
try {
Files.delete(p); // 删除文件或文件夹
} catch (IOException e) {
System.err.println("无法删除: " + p + ", 原因: " + e.getMessage());
}
});
}
} else {
logger.info("文件夹不存在: " + path);
}
} catch (Exception e) {
logger.error("deleteFolder", e);
}catch (Exception e) {
logger.error("deleteFolder",e);
}
}
public static List<String> readFiles(String stringPath) {
List<String> list = new ArrayList();
List<String> list = new ArrayList<>();
File directory = new File(stringPath);
// 检查目录是否存在且是一个目录
if (directory.exists() && directory.isDirectory()) {
// 获取目录下的所有子文件和文件夹
File[] files = directory.listFiles();
if (files != null) {
for(File file : files) {
for (File file : files) {
list.add(file.getName());
}
}
}
return list;
}
public static boolean isTempFile(String fileName) {
public static boolean isTempFile(String fileName) {
String[] result = fileName.split("\\.");
return result.length == 3;
}
public static boolean isFlv(String fileName) {
public static boolean isFlv(String fileName) {
return fileName.endsWith("flv");
}
public static boolean isMp4(String fileName) {
public static boolean isMp4(String fileName) {
return fileName.endsWith("mp4");
}
public static StreamType getStreamType(String fileName) {
if (fileName.endsWith("outer")) {
if(fileName.endsWith("outer")){
return StreamType.Outer;
} else {
return fileName.endsWith("inner") ? StreamType.Inner : StreamType.LiveVideo;
}else if(fileName.endsWith("inner")){
return StreamType.Inner;
}else{
return StreamType.LiveVideo;
}
}
}

121
src/main/java/com/tuoheng/steam/util/ProcessManager.java
View File

@ -1,127 +1,110 @@
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package com.tuoheng.steam.util;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import com.tuoheng.steam.controller.StreamRecordController;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.io.*;
import java.util.*;
@Service
public class ProcessManager {
private static final Logger logger = LoggerFactory.getLogger(ProcessManager.class);
private static final String DELIMITER = "=";
static Map<Long, String> runningProcessIds = new HashMap();
static String pidPath = "pid.txt";
public ProcessManager() {
}
static Map<Long, String> runningProcessIds = new HashMap<>();
public static void registerProcess(Long process) {
runningProcessIds.put(process, TimeUtils.formatDateToString(new Date()));
static String pidPath ="pid.txt";
static {
runningProcessIds = loadFromFile(pidPath);
for(Map.Entry<Long, String> entry : runningProcessIds.entrySet()){
try {
killProcessByPID(entry.getKey());
}catch (Exception e){
e.printStackTrace();
}
}
runningProcessIds.clear();
writeToFile(runningProcessIds);
}
public static void unRegisterProcess(Long process) {
public static void registerProcess(Long process) {
runningProcessIds.put(process,TimeUtils.formatDateToString(new Date()));
writeToFile(runningProcessIds);
}
public static void unRegisterProcess(Long process) {
runningProcessIds.remove(process);
writeToFile(runningProcessIds);
}
public static void writeToFile(Map<Long, String> map) {
public static void writeToFile(Map<Long, String> map){
try (BufferedWriter writer = new BufferedWriter(new FileWriter(pidPath))) {
for(Map.Entry<Long, String> entry : map.entrySet()) {
Object var10001 = entry.getKey();
writer.write(var10001 + "=" + (String)entry.getValue());
for (Map.Entry<Long, String> entry : map.entrySet()) {
writer.write(entry.getKey() + DELIMITER + entry.getValue());
writer.newLine();
}
} catch (IOException var6) {
}
}catch (IOException ignored){}
}
public static void appendToFile(Long pid, String strDate) {
try (BufferedWriter writer = new BufferedWriter(new FileWriter(pidPath, true))) {
writer.write(pid + "=" + strDate);
writer.newLine();
} catch (IOException var7) {
}
public static void appendToFile(Long pid,String strDate) {
try (BufferedWriter writer = new BufferedWriter(new FileWriter(pidPath,true))) {
writer.write(pid+ DELIMITER + strDate);
writer.newLine();
}catch (IOException ignored) {}
}
public static Map<Long, String> loadFromFile(String filePath) {
Map<Long, String> map = new LinkedHashMap();
public static Map<Long, String> loadFromFile(String filePath){
Map<Long, String> map = new LinkedHashMap<>();
File file = new File(filePath);
try {
if (!file.exists()) {
if(!file.exists()){
file.createNewFile();
}
} catch (Exception e) {
}catch (Exception e){
e.printStackTrace();
}
try {
try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
String line;
try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
while((line = reader.readLine()) != null) {
String[] parts = line.split("=", 2);
if (parts.length == 2) {
Long key = Long.parseLong(parts[0].trim());
String value = parts[1].trim();
map.put(key, value);
}
while ((line = reader.readLine()) != null) {
String[] parts = line.split(ProcessManager.DELIMITER, 2); // Split into key and value
if (parts.length == 2) {
Long key = Long.parseLong(parts[0].trim());
String value = parts[1].trim();
map.put(key, value);
}
}
return map;
} catch (IOException e) {
throw new RuntimeException(e);
}
return map;
}
public static void killProcessByPID(long pid) throws IOException, InterruptedException {
String os = System.getProperty("os.name").toLowerCase();
ProcessBuilder processBuilder;
if (os.contains("win")) {
processBuilder = new ProcessBuilder(new String[]{"cmd", "/c", "taskkill /PID " + pid + " /F"});
// Windows: Use taskkill command
processBuilder = new ProcessBuilder("cmd", "/c", "taskkill /PID " + pid + " /F");
} else {
processBuilder = new ProcessBuilder(new String[]{"bash", "-c", "kill -9 " + pid});
// Linux/Unix/macOS: Use kill command
processBuilder = new ProcessBuilder("bash", "-c", "kill -9 " + pid);
}
Process process = processBuilder.start();
int exitCode = process.waitFor();
if (exitCode == 0) {
logger.info("Process with PID " + pid + " terminated successfully.");
} else {
logger.error("Failed to terminate process with PID " + pid + ". Exit code: " + exitCode);
}
}
static {
runningProcessIds = loadFromFile(pidPath);
for(Map.Entry<Long, String> entry : runningProcessIds.entrySet()) {
try {
killProcessByPID((Long)entry.getKey());
} catch (Exception e) {
e.printStackTrace();
}
}
runningProcessIds.clear();
writeToFile(runningProcessIds);
}
}

22
src/main/resources/application.properties
View File

@ -7,8 +7,10 @@
#recordPath=/data/java/srs/srs/trunk/objs/nginx/html/record
#livedates=8
#cangneiwai=false
#大数据局生产
#srs.domain = ""
#srs.name = ""
#
#大数据局
#spring.application.name=stream_server
#server.port = 8989
#srs.splitPath=/data/java/srs/stream_server/temp
@ -17,10 +19,8 @@
#recordPath=/data/java/srs/srs/trunk/objs/nginx/html/record
#livedates=8
#cangneiwai=true
# 录制码率配置 (单位: kbps, 2000k = 2Mbps)
video.bitrate=4000k
audio.bitrate=128k
#srs.domain = ""
#srs.name = ""
#本地测试
#spring.application.name=stream_server
@ -31,9 +31,13 @@ audio.bitrate=128k
#recordPath=/Users/sunpeng/workspace/stream/record
#livedates=7
#cangneiwai=false
#srs.domain=
#srs.name=
#大数据容器化
##通过注入
##容器化部署
#通过注入
srs.domain = STREAM.t-aaron.com
#通过注入
srs.name = STREAM
spring.application.name=stream_server
server.port = 8080
@ -45,4 +49,4 @@ ffmpeg=ffmpeg
#
recordPath=/data/record
livedates=8
cangneiwai=false
cangneiwai=true