Skip to content

Commit 6d0ad06

Browse files
FireMasterKFireMasterK
authored
Add an alternative way for updating feeds for closed off instances (#889)
* feat(feeds): add DISABLE_PUBSUB to turn off WebSub subscriptions Gates the lease-renewal timer, the new-channel registration timer, and the subscribePubSub call in saveChannel, for a clean immediate disable without touching DISABLE_TIMERS or the cleanup timers. Reads the DISABLE_PUBSUB env var / config, defaults to false. * feat(feeds): add FEED_REFRESH periodic channel-refresh job A self-paced virtual-thread loop refreshes all users_subscribed channels over a configurable window (FEED_REFRESH on/off, FEED_REFRESH_MINUTES default 15), one channel every period/count. Extracts channelResponse's side effects into shared ChannelHelpers utils (videosTabInfo, federateChannelVideos, federateChannelInfo, updateChannelVideos) used by both the handler and the loop. --------- Co-authored-by: LogicalKarma <>
2 parents d2eb575 + e90776f commit 6d0ad06

5 files changed

Lines changed: 176 additions & 96 deletions

File tree

‎src/main/java/me/kavin/piped/Main.java‎

Lines changed: 51 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import org.hibernate.Session;
1919
import org.hibernate.StatelessSession;
2020
import org.schabi.newpipe.extractor.NewPipe;
21+
import org.schabi.newpipe.extractor.channel.ChannelInfo;
2122
import org.schabi.newpipe.extractor.localization.ContentCountry;
2223
import org.schabi.newpipe.extractor.localization.Localization;
2324
import org.schabi.newpipe.extractor.services.youtube.YoutubeJavaScriptPlayerManager;
@@ -122,7 +123,7 @@ public void run() {
122123
if (Constants.DISABLE_TIMERS)
123124
return;
124125

125-
new Timer().scheduleAtFixedRate(new TimerTask() {
126+
if (!Constants.DISABLE_PUBSUB) new Timer().scheduleAtFixedRate(new TimerTask() {
126127
@Override
127128
public void run() {
128129
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
@@ -181,7 +182,7 @@ public void run() {
181182
}
182183
}, 0, TimeUnit.MINUTES.toMillis(90));
183184

184-
new Timer().scheduleAtFixedRate(new TimerTask() {
185+
if (!Constants.DISABLE_PUBSUB) new Timer().scheduleAtFixedRate(new TimerTask() {
185186
@Override
186187
public void run() {
187188
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
@@ -257,5 +258,53 @@ public void run() {
257258
}
258259
}, 0, TimeUnit.MINUTES.toMillis(60));
259260

261+
if (Constants.FEED_REFRESH)
262+
Thread.ofVirtual().name("feed-refresh").start(() -> {
263+
while (true) {
264+
try {
265+
List<String> channelIds;
266+
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
267+
channelIds = s.createNativeQuery("SELECT DISTINCT channel FROM users_subscribed", String.class)
268+
.stream()
269+
.filter(Objects::nonNull)
270+
.distinct()
271+
.collect(Collectors.toCollection(ObjectArrayList::new));
272+
}
273+
274+
long periodMs = TimeUnit.MINUTES.toMillis(Constants.FEED_REFRESH_MINUTES);
275+
long delay = channelIds.isEmpty() ? periodMs : Math.max(1, periodMs / channelIds.size());
276+
277+
System.out.println("FeedRefresh: " + channelIds.size() + " channels, one every " + delay + "ms");
278+
279+
for (String channelId : channelIds) {
280+
try {
281+
var info = ChannelInfo.getInfo("https://youtube.com/channel/" + channelId);
282+
var tabInfo = ChannelHelpers.videosTabInfo(info);
283+
if (tabInfo != null)
284+
Multithreading.runAsync(() -> ChannelHelpers.federateChannelVideos(tabInfo));
285+
Multithreading.runAsync(() -> ChannelHelpers.federateChannelInfo(info));
286+
if (tabInfo != null)
287+
ChannelHelpers.updateChannelVideos(info, tabInfo);
288+
} catch (Exception e) {
289+
ExceptionHandler.handle(e);
290+
}
291+
Thread.sleep(delay);
292+
}
293+
294+
if (channelIds.isEmpty())
295+
Thread.sleep(periodMs);
296+
} catch (InterruptedException e) {
297+
break;
298+
} catch (Exception e) {
299+
ExceptionHandler.handle(e);
300+
try {
301+
Thread.sleep(TimeUnit.MINUTES.toMillis(1));
302+
} catch (InterruptedException ie) {
303+
break;
304+
}
305+
}
306+
}
307+
});
308+
260309
}
261310
}

‎src/main/java/me/kavin/piped/consts/Constants.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,12 @@ public class Constants {
6666

6767
public static final boolean DISABLE_TIMERS;
6868

69+
public static final boolean DISABLE_PUBSUB;
70+
71+
public static final boolean FEED_REFRESH;
72+
73+
public static final int FEED_REFRESH_MINUTES;
74+
6975
public static final String RYD_PROXY_URL;
7076

7177
public static final List<String> SPONSORBLOCK_SERVERS;
@@ -144,6 +150,9 @@ public class Constants {
144150
DISABLE_REGISTRATION = Boolean.parseBoolean(getProperty(prop, "DISABLE_REGISTRATION", "false"));
145151
FEED_RETENTION = Integer.parseInt(getProperty(prop, "FEED_RETENTION", "30"));
146152
DISABLE_TIMERS = Boolean.parseBoolean(getProperty(prop, "DISABLE_TIMERS", "false"));
153+
DISABLE_PUBSUB = Boolean.parseBoolean(getProperty(prop, "DISABLE_PUBSUB", "false"));
154+
FEED_REFRESH = Boolean.parseBoolean(getProperty(prop, "FEED_REFRESH", "false"));
155+
FEED_REFRESH_MINUTES = Integer.parseInt(getProperty(prop, "FEED_REFRESH_MINUTES", "15"));
147156
RYD_PROXY_URL = getProperty(prop, "RYD_PROXY_URL", "https://ryd-proxy.kavin.rocks");
148157
SPONSORBLOCK_SERVERS = List.of(getProperty(prop, "SPONSORBLOCK_SERVERS", "https://sponsor.ajay.app,https://sponsorblock.kavin.rocks")
149158
.split(","));

‎src/main/java/me/kavin/piped/server/handlers/ChannelHandlers.java‎

Lines changed: 4 additions & 93 deletions
Original file line numberDiff line numberDiff line change
@@ -41,106 +41,17 @@ public static byte[] channelResponse(String channelPath) throws Exception {
4141

4242
final ChannelInfo info = ChannelInfo.getInfo("https://youtube.com/" + channelPath);
4343

44-
final var preloadedVideosTab = collectPreloadedTabs(info.getTabs())
45-
.stream()
46-
.filter(tab -> tab.getContentFilters().contains(ChannelTabs.VIDEOS))
47-
.findFirst();
48-
49-
final ChannelTabInfo tabInfo = preloadedVideosTab.isPresent() ?
50-
ChannelTabInfo.getInfo(YOUTUBE_SERVICE, preloadedVideosTab.get()) :
51-
null;
44+
final ChannelTabInfo tabInfo = ChannelHelpers.videosTabInfo(info);
5245

5346
final List<ContentItem> relatedStreams = tabInfo != null ? collectRelatedItems(tabInfo.getRelatedItems()) : List.of();
5447

5548
if (tabInfo != null)
56-
Multithreading.runAsync(() -> tabInfo.getRelatedItems()
57-
.stream().filter(StreamInfoItem.class::isInstance)
58-
.map(StreamInfoItem.class::cast)
59-
.forEach(infoItem -> {
60-
if (
61-
infoItem.getUploadDate() != null &&
62-
System.currentTimeMillis() - infoItem.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
63-
< TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION)
64-
)
65-
try {
66-
MatrixHelper.sendEvent("video.piped.stream.info", new FederatedVideoInfo(
67-
StringUtils.substring(infoItem.getUrl(), -11), StringUtils.substring(infoItem.getUploaderUrl(), -24),
68-
infoItem.getName(),
69-
infoItem.getDuration(), infoItem.getViewCount())
70-
);
71-
} catch (IOException e) {
72-
throw new RuntimeException(e);
73-
}
74-
})
75-
);
49+
Multithreading.runAsync(() -> ChannelHelpers.federateChannelVideos(tabInfo));
7650

77-
Multithreading.runAsync(() -> {
78-
try {
79-
MatrixHelper.sendEvent("video.piped.channel.info", new FederatedChannelInfo(
80-
info.getId(), StringUtils.abbreviate(info.getName(), 100), info.getAvatars().isEmpty() ? null : info.getAvatars().getLast().getUrl(), info.isVerified())
81-
);
82-
} catch (IOException e) {
83-
throw new RuntimeException(e);
84-
}
85-
});
51+
Multithreading.runAsync(() -> ChannelHelpers.federateChannelInfo(info));
8652

8753
if (tabInfo != null)
88-
Multithreading.runAsync(() -> {
89-
90-
var channel = DatabaseHelper.getChannelFromId(info.getId());
91-
92-
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
93-
94-
if (channel != null) {
95-
96-
ChannelHelpers.updateChannel(s, channel, StringUtils.abbreviate(info.getName(), 100), info.getAvatars().isEmpty() ? null : info.getAvatars().getLast().getUrl(), info.isVerified());
97-
98-
Set<String> ids = tabInfo.getRelatedItems()
99-
.stream()
100-
.filter(StreamInfoItem.class::isInstance)
101-
.map(StreamInfoItem.class::cast)
102-
.filter(item -> {
103-
long time = item.getUploadDate() != null
104-
? item.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
105-
: System.currentTimeMillis();
106-
return System.currentTimeMillis() - time < TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION);
107-
})
108-
.map(item -> {
109-
try {
110-
return YOUTUBE_SERVICE.getStreamLHFactory().getId(item.getUrl());
111-
} catch (ParsingException e) {
112-
throw new RuntimeException(e);
113-
}
114-
})
115-
.collect(Collectors.toUnmodifiableSet());
116-
117-
List<Video> videos = DatabaseHelper.getVideosFromIds(s, ids);
118-
119-
tabInfo.getRelatedItems()
120-
.stream()
121-
.filter(StreamInfoItem.class::isInstance)
122-
.map(StreamInfoItem.class::cast).forEach(item -> {
123-
long time = item.getUploadDate() != null
124-
? item.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
125-
: System.currentTimeMillis();
126-
if (System.currentTimeMillis() - time < TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION))
127-
try {
128-
String id = YOUTUBE_SERVICE.getStreamLHFactory().getId(item.getUrl());
129-
var video = videos.stream()
130-
.filter(v -> v.getId().equals(id))
131-
.findFirst();
132-
if (video.isPresent()) {
133-
VideoHelpers.updateVideo(id, item);
134-
} else {
135-
VideoHelpers.handleNewVideo("https://youtube.com/watch?v=" + id, time, channel);
136-
}
137-
} catch (Exception e) {
138-
ExceptionHandler.handle(e);
139-
}
140-
});
141-
}
142-
}
143-
});
54+
Multithreading.runAsync(() -> ChannelHelpers.updateChannelVideos(info, tabInfo));
14455

14556
String nextpage = null;
14657
if (tabInfo != null && tabInfo.hasNextPage()) {

‎src/main/java/me/kavin/piped/utils/ChannelHelpers.java‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,22 @@
77
import com.rometools.modules.mediarss.types.Thumbnail;
88
import com.rometools.rome.feed.synd.*;
99
import me.kavin.piped.consts.Constants;
10+
import me.kavin.piped.utils.obj.MatrixHelper;
1011
import me.kavin.piped.utils.obj.db.Channel;
1112
import me.kavin.piped.utils.obj.db.Video;
13+
import me.kavin.piped.utils.obj.federation.FederatedChannelInfo;
14+
import me.kavin.piped.utils.obj.federation.FederatedVideoInfo;
1215
import okhttp3.Request;
1316
import org.apache.commons.lang3.StringUtils;
1417
import org.apache.commons.lang3.time.DurationFormatUtils;
1518
import org.apache.commons.text.StringEscapeUtils;
1619
import org.hibernate.StatelessSession;
20+
import org.schabi.newpipe.extractor.channel.ChannelInfo;
21+
import org.schabi.newpipe.extractor.channel.tabs.ChannelTabInfo;
22+
import org.schabi.newpipe.extractor.channel.tabs.ChannelTabs;
23+
import org.schabi.newpipe.extractor.exceptions.ExtractionException;
24+
import org.schabi.newpipe.extractor.exceptions.ParsingException;
25+
import org.schabi.newpipe.extractor.stream.StreamInfoItem;
1726

1827
import java.io.IOException;
1928
import java.net.MalformedURLException;
@@ -22,7 +31,12 @@
2231
import java.util.Collections;
2332
import java.util.Date;
2433
import java.util.List;
34+
import java.util.Set;
35+
import java.util.concurrent.TimeUnit;
36+
import java.util.stream.Collectors;
2537

38+
import static me.kavin.piped.consts.Constants.YOUTUBE_SERVICE;
39+
import static me.kavin.piped.utils.CollectionUtils.collectPreloadedTabs;
2640
import static me.kavin.piped.utils.URLUtils.rewriteURL;
2741

2842
public class ChannelHelpers {
@@ -75,6 +89,103 @@ public static void updateChannel(StatelessSession s, Channel channel, String nam
7589
}
7690
}
7791

92+
public static void updateChannelVideos(ChannelInfo info, ChannelTabInfo tabInfo) {
93+
94+
var channel = DatabaseHelper.getChannelFromId(info.getId());
95+
96+
try (StatelessSession s = DatabaseSessionFactory.createStatelessSession()) {
97+
98+
if (channel != null) {
99+
100+
updateChannel(s, channel, StringUtils.abbreviate(info.getName(), 100), info.getAvatars().isEmpty() ? null : info.getAvatars().getLast().getUrl(), info.isVerified());
101+
102+
Set<String> ids = tabInfo.getRelatedItems()
103+
.stream()
104+
.filter(StreamInfoItem.class::isInstance)
105+
.map(StreamInfoItem.class::cast)
106+
.filter(item -> {
107+
long time = item.getUploadDate() != null
108+
? item.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
109+
: System.currentTimeMillis();
110+
return System.currentTimeMillis() - time < TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION);
111+
})
112+
.map(item -> {
113+
try {
114+
return YOUTUBE_SERVICE.getStreamLHFactory().getId(item.getUrl());
115+
} catch (ParsingException e) {
116+
throw new RuntimeException(e);
117+
}
118+
})
119+
.collect(Collectors.toUnmodifiableSet());
120+
121+
List<Video> videos = DatabaseHelper.getVideosFromIds(s, ids);
122+
123+
tabInfo.getRelatedItems()
124+
.stream()
125+
.filter(StreamInfoItem.class::isInstance)
126+
.map(StreamInfoItem.class::cast).forEach(item -> {
127+
long time = item.getUploadDate() != null
128+
? item.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
129+
: System.currentTimeMillis();
130+
if (System.currentTimeMillis() - time < TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION))
131+
try {
132+
String id = YOUTUBE_SERVICE.getStreamLHFactory().getId(item.getUrl());
133+
var video = videos.stream()
134+
.filter(v -> v.getId().equals(id))
135+
.findFirst();
136+
if (video.isPresent()) {
137+
VideoHelpers.updateVideo(id, item);
138+
} else {
139+
VideoHelpers.handleNewVideo("https://youtube.com/watch?v=" + id, time, channel);
140+
}
141+
} catch (Exception e) {
142+
ExceptionHandler.handle(e);
143+
}
144+
});
145+
}
146+
}
147+
}
148+
149+
public static ChannelTabInfo videosTabInfo(ChannelInfo info) throws ExtractionException, IOException {
150+
var preloadedVideosTab = collectPreloadedTabs(info.getTabs())
151+
.stream()
152+
.filter(tab -> tab.getContentFilters().contains(ChannelTabs.VIDEOS))
153+
.findFirst();
154+
return preloadedVideosTab.isPresent() ? ChannelTabInfo.getInfo(YOUTUBE_SERVICE, preloadedVideosTab.get()) : null;
155+
}
156+
157+
public static void federateChannelVideos(ChannelTabInfo tabInfo) {
158+
tabInfo.getRelatedItems()
159+
.stream().filter(StreamInfoItem.class::isInstance)
160+
.map(StreamInfoItem.class::cast)
161+
.forEach(infoItem -> {
162+
if (
163+
infoItem.getUploadDate() != null &&
164+
System.currentTimeMillis() - infoItem.getUploadDate().offsetDateTime().toInstant().toEpochMilli()
165+
< TimeUnit.DAYS.toMillis(Constants.FEED_RETENTION)
166+
)
167+
try {
168+
MatrixHelper.sendEvent("video.piped.stream.info", new FederatedVideoInfo(
169+
StringUtils.substring(infoItem.getUrl(), -11), StringUtils.substring(infoItem.getUploaderUrl(), -24),
170+
infoItem.getName(),
171+
infoItem.getDuration(), infoItem.getViewCount())
172+
);
173+
} catch (IOException e) {
174+
throw new RuntimeException(e);
175+
}
176+
});
177+
}
178+
179+
public static void federateChannelInfo(ChannelInfo info) {
180+
try {
181+
MatrixHelper.sendEvent("video.piped.channel.info", new FederatedChannelInfo(
182+
info.getId(), StringUtils.abbreviate(info.getName(), 100), info.getAvatars().isEmpty() ? null : info.getAvatars().getLast().getUrl(), info.isVerified())
183+
);
184+
} catch (IOException e) {
185+
throw new RuntimeException(e);
186+
}
187+
}
188+
78189
public static SyndEntry createEntry(Video video, Channel channel) {
79190
SyndEntry entry = new SyndEntryImpl();
80191
SyndPerson person = new SyndPersonImpl();

‎src/main/java/me/kavin/piped/utils/DatabaseHelper.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -202,7 +202,7 @@ public static Channel saveChannel(String channelId) {
202202
ExceptionHandler.handle(e);
203203
}
204204

205-
Multithreading.runAsync(() -> {
205+
if (!Constants.DISABLE_PUBSUB) Multithreading.runAsync(() -> {
206206
try {
207207
PubSubHelper.subscribePubSub(channelId);
208208
} catch (IOException e) {

0 commit comments

Comments
 (0)