TeamPiped-mirror/Piped-Backend
1
0
Fork 0
mirror of https://github.com/TeamPiped/Piped-Backend.git synced 2024-08-14 23:51:41 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #642 from TeamPiped/pubsub-upsert

Browse source 
Use upsert for handling pubsub and run on limited executors
This commit is contained in:
Kavin 2023-07-06 15:02:00 +01:00 committed by GitHub
commit f99aea64cf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 14 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
src/main/java/me/kavin/piped/server/ServerLauncher.java
View file

@ -76,18 +76,21 @@ public class ServerLauncher extends MultithreadedHttpServerLauncher {
.map(GET, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> { .map(GET, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> {
var topic = request.getQueryParameter("hub.topic"); var topic = request.getQueryParameter("hub.topic");
if (topic != null) if (topic != null)
Multithreading.runAsync(() -> { Multithreading.runAsyncLimited(() -> {
String channelId = StringUtils.substringAfter(topic, "channel_id="); String channelId = StringUtils.substringAfter(topic, "channel_id=");
PubSubHelper.updatePubSub(channelId); PubSubHelper.updatePubSub(channelId);
}); });
return HttpResponse.ok200().withPlainText(Objects.requireNonNull(request.getQueryParameter("hub.challenge")));
var challenge = request.getQueryParameter("hub.challenge");
return HttpResponse.ok200()
.withPlainText(Objects.requireNonNullElse(challenge, "ok"));
})).map(POST, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> { })).map(POST, "/webhooks/pubsub", AsyncServlet.ofBlocking(executor, request -> {
try { try {
SyndFeed feed = new SyndFeedInput().build( SyndFeed feed = new SyndFeedInput().build(
new InputSource(new ByteArrayInputStream(request.loadBody().getResult().asArray()))); new InputSource(new ByteArrayInputStream(request.loadBody().getResult().asArray())));
Multithreading.runAsync(() -> { Multithreading.runAsyncLimited(() -> {
for (var entry : feed.getEntries()) { for (var entry : feed.getEntries()) {
String url = entry.getLinks().get(0).getHref(); String url = entry.getLinks().get(0).getHref();
String videoId = StringUtils.substring(url, -11); String videoId = StringUtils.substring(url, -11);
@ -95,7 +98,7 @@ public class ServerLauncher extends MultithreadedHttpServerLauncher {
if (DatabaseHelper.doesVideoExist(s, videoId)) if (DatabaseHelper.doesVideoExist(s, videoId))
continue; continue;
} }
Multithreading.runAsync(() -> { Multithreading.runAsyncLimited(() -> {
try { try {
Sentry.setExtra("videoId", videoId); Sentry.setExtra("videoId", videoId);
var extractor = YOUTUBE_SERVICE.getStreamExtractor("https://youtube.com/watch?v=" + videoId); var extractor = YOUTUBE_SERVICE.getStreamExtractor("https://youtube.com/watch?v=" + videoId);

17
src/main/java/me/kavin/piped/utils/PubSubHelper.java
View file

@ -57,17 +57,14 @@ public class PubSubHelper {
} }
public static void updatePubSub(String channelId) { public static void updatePubSub(String channelId) {
var pubsub = DatabaseHelper.getPubSubFromId(channelId);
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) { try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
s.beginTransaction(); var tr = s.beginTransaction();
if (pubsub == null) { s.createNativeMutationQuery("INSERT INTO pubsub (id, subbed_at) VALUES (?, ?) " +
pubsub = new PubSub(channelId, System.currentTimeMillis()); "ON CONFLICT (id) DO UPDATE SET subbed_at = excluded.subbed_at")
s.insert(pubsub); .setParameter(1, channelId)
} else { .setParameter(2, System.currentTimeMillis())
pubsub.setSubbedAt(System.currentTimeMillis()); .executeUpdate();
s.update(pubsub); tr.commit();
}
s.getTransaction().commit();
} }
} }
} }