davebluke
/
ctbrec
forked from j62/ctbrec
1
0
Fork 0
Code Pull Requests Activity

Add a timeout of 2 seconds for each online check to make sure the online check doesn't get blocked somehow

This commit is contained in:
0xb00bface 2022-04-10 17:47:40 +02:00
parent a008fff084
commit 04eb5a7ad1
3 changed files with 191 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/src/main/resources/logback.xml
View File

@ -45,7 +45,7 @@
<logger name="ctbrec.LoggingInterceptor" level="INFO"/>
<logger name="ctbrec.io.CookieJarImpl" level="INFO"/>
<logger name="ctbrec.recorder.FFmpeg" level="DEBUG"/>
<logger name="ctbrec.recorder.OnlineMonitor" level="INFO"/>
<logger name="ctbrec.recorder.OnlineMonitor" level="DEBUG"/>
<logger name="ctbrec.recorder.RecordingFileMonitor" level="TRACE"/>
<logger name="ctbrec.recorder.download.dash.DashDownload" level="DEBUG"/>
<logger name="ctbrec.recorder.server.HlsServlet" level="INFO"/>

106
common/src/main/java/ctbrec/recorder/OnlineMonitor.java
View File

@ -1,44 +1,38 @@
package ctbrec.recorder;
import static ctbrec.Model.State.*;
import java.io.InterruptedIOException;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.time.Instant;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ctbrec.Config;
import ctbrec.Model;
import ctbrec.event.EventBusHolder;
import ctbrec.event.ModelIsOnlineEvent;
import ctbrec.event.ModelStateChangedEvent;
import ctbrec.io.HttpException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InterruptedIOException;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.time.Instant;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import static ctbrec.Model.State.UNKNOWN;
public class OnlineMonitor extends Thread {
private static final Logger LOG = LoggerFactory.getLogger(OnlineMonitor.class);
private static final boolean IGNORE_CACHE = true;
private static final long TIMEOUT_IN_MILLIS = 2000;
private volatile boolean running = false;
private Recorder recorder;
private final Recorder recorder;
private Map<Model, Model.State> states = new HashMap<>();
private final Map<Model, Model.State> states = new HashMap<>();
private Map<String, ExecutorService> executors = new HashMap<>();
private Config config;
private final Map<String, ExecutorService> executors = new HashMap<>();
private final Config config;
public OnlineMonitor(Recorder recorder, Config config) {
this.recorder = recorder;
@ -68,19 +62,14 @@ public class OnlineMonitor extends Thread {
}
private void removeDeletedModels(List<Model> models) {
for (Iterator<Model> iterator = states.keySet().iterator(); iterator.hasNext();) {
Model model = iterator.next();
if(!models.contains(model)) {
iterator.remove();
}
}
states.keySet().removeIf(model -> !models.contains(model));
}
private void updateModels(List<Model> models) {
// sort models by priority
Collections.sort(models, (a, b) -> b.getPriority() - a.getPriority());
models.sort((a, b) -> b.getPriority() - a.getPriority());
// submit online check jobs to the executor for the model's site
List<Future<?>> futures = new LinkedList<>();
List<ModelAwareFuture> futures = new LinkedList<>();
for (Model model : models) {
boolean skipCheckForSuspended = config.getSettings().onlineCheckSkipsPausedModels && model.isSuspended();
boolean skipCheckForMarkedAsLater = model.isMarkedForLaterRecording();
@ -91,18 +80,21 @@ public class OnlineMonitor extends Thread {
}
}
// wait for all jobs to finish
for (Future<?> future : futures) {
for (ModelAwareFuture future : futures) {
try {
future.get();
future.get(TIMEOUT_IN_MILLIS, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.debug("Online check interrupted for model {}", future.getModel(), e);
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
LOG.info("Error while checking online state", e);
LOG.info("Error while checking online state for model {}", future.getModel(), e);
} catch (TimeoutException e) {
LOG.debug("Online check didn't finish after {}ms for model {}", TIMEOUT_IN_MILLIS, future.getModel());
}
}
}
private Future<?> updateModel(Model model) {
private ModelAwareFuture updateModel(Model model) {
final String siteName = model.getSite().getName();
ExecutorService executor = executors.computeIfAbsent(siteName, name -> Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r);
@ -112,14 +104,14 @@ public class OnlineMonitor extends Thread {
return t;
}));
return executor.submit(() -> {
var future = executor.submit(() -> {
try {
if (model.isOnline(IGNORE_CACHE)) {
EventBusHolder.BUS.post(new ModelIsOnlineEvent(model));
model.setLastSeen(Instant.now());
}
Model.State state = model.getOnlineState(false);
LOG.trace("Model online state: {} {}", model.getName(), state);
LOG.debug("Model online state: {} {}", model.getName(), state);
Model.State oldState = states.getOrDefault(model, UNKNOWN);
states.put(model, state);
if (state != oldState) {
@ -137,13 +129,15 @@ public class OnlineMonitor extends Thread {
} catch (Exception e) {
LOG.error("Couldn't check if model {} is online", model.getName(), e);
}
return model;
});
return new ModelAwareFuture(model, future);
}
private void suspendUntilNextIteration(List<Model> models, Duration timeCheckTook) {
LOG.debug("Online check for {} models took {} seconds", models.size(), timeCheckTook.getSeconds());
long sleepTime = config.getSettings().onlineCheckIntervalInSecs;
if(timeCheckTook.getSeconds() < sleepTime) {
if (timeCheckTook.getSeconds() < sleepTime) {
try {
if (running) {
long millis = TimeUnit.SECONDS.toMillis(sleepTime - timeCheckTook.getSeconds());
@ -164,4 +158,36 @@ public class OnlineMonitor extends Thread {
}
interrupt();
}
private record ModelAwareFuture(Model model, Future<Model> future) implements Future<Model> {
public Model getModel() {
return model;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return future.cancel(mayInterruptIfRunning);
}
@Override
public boolean isCancelled() {
return future.isCancelled();
}
@Override
public boolean isDone() {
return future.isDone();
}
@Override
public Model get() throws InterruptedException, ExecutionException {
return future.get();
}
@Override
public Model get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
return future.get(timeout, unit);
}
}
}

19
common/src/main/java/ctbrec/sites/mfc/MyFreeCamsClient.java
View File

@ -20,6 +20,7 @@ import java.net.URLDecoder;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
@ -125,8 +126,8 @@ public class MyFreeCamsClient {
.header(USER_AGENT, Config.getInstance().getSettings().httpUserAgent)
.header(CONNECTION, KEEP_ALIVE)
.build();
try(Response resp = mfc.getHttpClient().execute(req)) {
if(!resp.isSuccessful()) {
try (Response resp = mfc.getHttpClient().execute(req)) {
if (!resp.isSuccessful()) {
throw new HttpException(resp.code(), resp.message());
}
}
@ -598,7 +599,7 @@ public class MyFreeCamsClient {
for (SessionState state : sessionStates.asMap().values()) {
Optional<String> nm = Optional.ofNullable(state.getNm());
Optional<String> name = Optional.ofNullable(model.getName());
if(nm.isEmpty() || name.isEmpty()) {
if (nm.isEmpty() || name.isEmpty()) {
continue;
}
@ -672,7 +673,9 @@ public class MyFreeCamsClient {
LOG.trace("Sending USERNAMELOOKUP for {}", q);
Object monitor = new Object();
int msgId = messageId++;
AtomicBoolean searchDone = new AtomicBoolean(false);
responseHandlers.put(msgId, msg -> {
try {
LOG.trace("Search result: {}", msg);
if (StringUtil.isNotBlank(msg.getMessage()) && !Objects.equals(msg.getMessage(), q)) {
JSONObject json = new JSONObject(msg.getMessage());
@ -695,13 +698,21 @@ public class MyFreeCamsClient {
model.setPreview(previewUrl);
result.add(model);
}
} finally {
searchDone.setPlain(true);
synchronized (monitor) {
monitor.notifyAll();
}
}
});
ws.send("10 " + sessionId + " 0 " + msgId + " 0 " + q + "\n");
int waitInMillis = 1000;
int iterations = 0;
while (iterations < 5 && !searchDone.get()) {
synchronized (monitor) {
monitor.wait();
monitor.wait(waitInMillis);
iterations++;
}
}
for (MyFreeCamsModel model : models.asMap().values()) {