diff --git a/common/src/main/java/ctbrec/recorder/NextGenLocalRecorder.java b/common/src/main/java/ctbrec/recorder/NextGenLocalRecorder.java index a1ed85ff..acc60203 100644 --- a/common/src/main/java/ctbrec/recorder/NextGenLocalRecorder.java +++ b/common/src/main/java/ctbrec/recorder/NextGenLocalRecorder.java @@ -101,16 +101,31 @@ public class NextGenLocalRecorder implements Recorder { try { Future result = completionService.take(); Recording recording = result.get(); - recordingProcesses.remove(recording.getModel()); + recordingsLock.lock(); + try { + recordingProcesses.remove(recording.getModel()); + } finally { + recordingsLock.unlock(); + } if (recording.getStatus() == State.WAITING) { LOG.debug("Download finished for {} -> Starting post-processing", recording.getModel().getName()); ppPool.submit(() -> { - setRecordingStatus(recording, State.POST_PROCESSING); - recordingManager.saveRecording(recording); - recording.postprocess(); - setRecordingStatus(recording, State.FINISHED); - recordingManager.saveRecording(recording); - return recording; + try { + setRecordingStatus(recording, State.POST_PROCESSING); + recordingManager.saveRecording(recording); + recording.postprocess(); + setRecordingStatus(recording, State.FINISHED); + recordingManager.saveRecording(recording); + deleteIfTooShort(recording); + } catch (Exception e) { + LOG.error("Error while post-processing recording {}", recording, e); + recording.setStatus(State.FAILED); + try { + recordingManager.saveRecording(recording); + } catch (IOException e1) { + LOG.error("Couldn't update recording state for recording {}", recording, e1); + } + } }); // check, if we have to restart the recording @@ -118,19 +133,12 @@ public class NextGenLocalRecorder implements Recorder { tryRestartRecording(model); } else { if(recording.getStatus() != State.DELETED) { - recordingsLock.lock(); - try { - recordingManager.delete(recording); - } catch (IOException e) { - LOG.error("Couldn't delete recording {}", recording, e); - } finally { - recordingsLock.unlock(); - } + delete(recording); } setRecordingStatus(recording, State.FAILED); } - } catch (InterruptedException | ExecutionException e) { - e.printStackTrace(); + } catch (InterruptedException | ExecutionException | InvalidKeyException | NoSuchAlgorithmException | IllegalStateException | IOException e) { + LOG.error("Error while completing recording", e); } } }); @@ -231,7 +239,7 @@ public class NextGenLocalRecorder implements Recorder { Recording rec = new Recording(); rec.setDownload(download); - rec.setPath(download.getPath(model)); + rec.setPath(download.getPath(model).replaceAll("\\\\", "/")); rec.setModel(model); rec.setStartDate(Instant.ofEpochMilli(System.currentTimeMillis())); recordingProcesses.put(model, rec); @@ -241,7 +249,7 @@ public class NextGenLocalRecorder implements Recorder { setRecordingStatus(rec, State.RECORDING); recordingManager.saveRecording(rec); download.start(); - boolean deleted = deleteIfTooShort(rec); + boolean deleted = deleteIfEmpty(rec); setRecordingStatus(rec, deleted ? State.DELETED : State.WAITING); recordingManager.saveRecording(rec); } catch (IOException e) { @@ -254,15 +262,19 @@ public class NextGenLocalRecorder implements Recorder { } } - private boolean deleteIfTooShort(Recording rec) throws IOException, ParseException, PlaylistException { - // if the size is 0, we don't need to go ahead and check the length + private boolean deleteIfEmpty(Recording rec) throws IOException, ParseException, PlaylistException, InvalidKeyException, NoSuchAlgorithmException, IllegalStateException { rec.refresh(); long sizeInByte = rec.getSizeInByte(); if (sizeInByte == 0) { - recordingManager.delete(rec); + LOG.info("Deleting empty recording {}", rec); + delete(rec); return true; + } else { + return false; } + } + private boolean deleteIfTooShort(Recording rec) throws IOException, ParseException, PlaylistException, InvalidKeyException, NoSuchAlgorithmException, IllegalStateException { Duration minimumLengthInSeconds = Duration.ofSeconds(Config.getInstance().getSettings().minimumLengthInSeconds); if (minimumLengthInSeconds.getSeconds() <= 0) { return false; @@ -270,7 +282,8 @@ public class NextGenLocalRecorder implements Recorder { Duration recordingLength = rec.getLength(); if (recordingLength.compareTo(minimumLengthInSeconds) < 0) { - recordingManager.delete(rec); + LOG.info("Deleting too short recording {} [{} < {}]", rec, recordingLength, minimumLengthInSeconds); + delete(rec); return true; } @@ -293,9 +306,14 @@ public class NextGenLocalRecorder implements Recorder { modelLock.unlock(); } - if (recordingProcesses.containsKey(model)) { - Recording recording = recordingProcesses.get(model); - recording.getDownload().stop(); + recordingsLock.lock(); + try { + if (recordingProcesses.containsKey(model)) { + Recording recording = recordingProcesses.get(model); + recording.getDownload().stop(); + } + } finally { + recordingsLock.unlock(); } } @@ -306,9 +324,14 @@ public class NextGenLocalRecorder implements Recorder { models.get(index).setStreamUrlIndex(model.getStreamUrlIndex()); config.save(); LOG.debug("Switching stream source to index {} for model {}", model.getStreamUrlIndex(), model.getName()); - Recording recording = recordingProcesses.get(model); - if (recording != null) { - stopRecordingProcess(model); + recordingsLock.lock(); + try { + Recording recording = recordingProcesses.get(model); + if (recording != null) { + stopRecordingProcess(model); + } + } finally { + recordingsLock.unlock(); } tryRestartRecording(model); } else { @@ -318,10 +341,15 @@ public class NextGenLocalRecorder implements Recorder { } private void stopRecordingProcess(Model model) { - LOG.debug("Stopping recording for {}", model); - Recording recording = recordingProcesses.get(model); - LOG.debug("Stopping download for {}", model); - recording.getDownload().stop(); + recordingsLock.lock(); + try { + LOG.debug("Stopping recording for {}", model); + Recording recording = recordingProcesses.get(model); + LOG.debug("Stopping download for {}", model); + recording.getDownload().stop(); + } finally { + recordingsLock.unlock(); + } } private void stopRecordingProcesses() { @@ -357,12 +385,7 @@ public class NextGenLocalRecorder implements Recorder { @Override public List getRecordings() throws IOException, InvalidKeyException, NoSuchAlgorithmException, IllegalStateException { - recordingsLock.lock(); - try { - return recordingManager.getAll(); - } finally { - recordingsLock.unlock(); - } + return recordingManager.getAll(); } @Override @@ -377,20 +400,25 @@ public class NextGenLocalRecorder implements Recorder { recording = false; LOG.debug("Stopping all recording processes"); - // make a copy to avoid ConcurrentModificationException - List toStop = new ArrayList<>(recordingProcesses.values()); - for (Recording rec : toStop) { - Optional.ofNullable(rec.getDownload()).ifPresent(Download::stop); - } - - // wait for post-processing to finish - LOG.info("Waiting for downloads to finish"); - while (!recordingProcesses.isEmpty()) { - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - LOG.error("Error while waiting for downloads to finish", e); + recordingsLock.lock(); + try { + // make a copy to avoid ConcurrentModificationException + List toStop = new ArrayList<>(recordingProcesses.values()); + for (Recording rec : toStop) { + Optional.ofNullable(rec.getDownload()).ifPresent(Download::stop); } + + // wait for post-processing to finish + LOG.info("Waiting for downloads to finish"); + while (!recordingProcesses.isEmpty()) { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + LOG.error("Error while waiting for downloads to finish", e); + } + } + } finally { + recordingsLock.unlock(); } // shutdown threadpools @@ -426,8 +454,13 @@ public class NextGenLocalRecorder implements Recorder { modelLock.unlock(); } - Recording recording = recordingProcesses.get(model); - Optional.ofNullable(recording).map(Recording::getDownload).ifPresent(Download::stop); + recordingsLock.lock(); + try { + Recording recording = recordingProcesses.get(model); + Optional.ofNullable(recording).map(Recording::getDownload).ifPresent(Download::stop); + } finally { + recordingsLock.unlock(); + } } @Override @@ -541,6 +574,7 @@ public class NextGenLocalRecorder implements Recorder { EventBusHolder.BUS.register(new Object() { @Subscribe public void modelEvent(Event e) { + recordingsLock.lock(); try { if (e.getType() == MODEL_ONLINE) { ModelIsOnlineEvent evt = (ModelIsOnlineEvent) e; @@ -551,6 +585,8 @@ public class NextGenLocalRecorder implements Recorder { } } catch (Exception e1) { LOG.error("Error while handling model state changed event {}", e, e1); + } finally { + recordingsLock.unlock(); } } }); diff --git a/common/src/main/java/ctbrec/recorder/RecordingManager.java b/common/src/main/java/ctbrec/recorder/RecordingManager.java index 0517038d..666ae7ba 100644 --- a/common/src/main/java/ctbrec/recorder/RecordingManager.java +++ b/common/src/main/java/ctbrec/recorder/RecordingManager.java @@ -9,6 +9,7 @@ import java.nio.file.Files; import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.locks.ReentrantLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,7 +32,7 @@ public class RecordingManager { private Moshi moshi; private JsonAdapter adapter; private List recordings = new ArrayList<>(); - // private RecordingFileMonitor monitor = new RecordingFileMonitor(this); + private ReentrantLock recordingsLock = new ReentrantLock(); public RecordingManager(Config config, List sites) throws IOException { this.config = config; @@ -42,13 +43,16 @@ public class RecordingManager { adapter = moshi.adapter(Recording.class).indent(" "); loadRecordings(); - // startMonitoring(); } public void add(Recording rec) throws UnsupportedEncodingException, IOException { saveRecording(rec); - recordings.add(rec); - // registerFileWatch(rec); + recordingsLock.lock(); + try { + recordings.add(rec); + } finally { + recordingsLock.unlock(); + } } public void saveRecording(Recording rec) throws UnsupportedEncodingException, IOException { @@ -81,35 +85,6 @@ public class RecordingManager { } } - // private void startMonitoring() { - // for (Recording recording : recordings) { - // registerFileWatch(recording); - // } - // Thread watcher = new Thread(() -> monitor.processEvents()); - // watcher.setDaemon(true); - // watcher.setPriority(Thread.MIN_PRIORITY); - // watcher.setName("RecordingFileMonitor"); - // watcher.start(); - // } - // - // private void registerFileWatch(Recording recording) { - // File rec = recording.getAbsoluteFile(); - // if (rec.isDirectory()) { - // monitor.register(rec.toPath()); - // } else { - // monitor.register(rec.getParentFile().toPath()); - // } - // } - // - // private void removeFileWatch(Recording recording) { - // File rec = recording.getAbsoluteFile(); - // if (rec.isDirectory()) { - // monitor.unregister(rec.toPath()); - // } else { - // monitor.unregister(rec.getParentFile().toPath()); - // } - // } - private boolean recordingExists(Recording recording) { File rec = new File(config.getSettings().recordingsDir, recording.getPath()); return rec.exists(); @@ -122,38 +97,46 @@ public class RecordingManager { } public void delete(Recording recording) throws IOException { - int idx = recordings.indexOf(recording); - recording = recordings.get(idx); + recordingsLock.lock(); + try { + int idx = recordings.indexOf(recording); + recording = recordings.get(idx); - recording.setStatus(State.DELETING); - File recordingsDir = new File(config.getSettings().recordingsDir); - File path = new File(recordingsDir, recording.getPath()); - LOG.debug("Deleting {}", path); + recording.setStatus(State.DELETING); + File recordingsDir = new File(config.getSettings().recordingsDir); + File path = new File(recordingsDir, recording.getPath()); + LOG.debug("Deleting {}", path); - // delete the video files - if (path.isFile()) { - Files.delete(path.toPath()); - deleteEmptyParents(path.getParentFile()); - } else { - deleteDirectory(path); - deleteEmptyParents(path); + // delete the video files + if (path.isFile()) { + Files.delete(path.toPath()); + deleteEmptyParents(path.getParentFile()); + } else { + deleteDirectory(path); + deleteEmptyParents(path); + } + + // delete the meta data + Files.deleteIfExists(new File(recording.getMetaDataFile()).toPath()); + + // remove from data structure + recordings.remove(recording); + recording.setStatus(State.DELETED); + } finally { + recordingsLock.unlock(); } - - // delete the meta data - Files.deleteIfExists(new File(recording.getMetaDataFile()).toPath()); - - // remove from data structure - recordings.remove(recording); - recording.setStatus(State.DELETED); - - // removeFileWatch(recording); } public List getAll() { - for (Recording recording : recordings) { - recording.refresh(); + recordingsLock.lock(); + try { + for (Recording recording : recordings) { + recording.refresh(); + } + return new ArrayList<>(recordings); + } finally { + recordingsLock.unlock(); } - return new ArrayList<>(recordings); } private void deleteEmptyParents(File parent) throws IOException {