diff options
author | Chocobozzz <me@florianbigard.com> | 2022-06-17 14:08:13 +0200 |
---|---|---|
committer | Chocobozzz <me@florianbigard.com> | 2022-06-17 14:08:13 +0200 |
commit | f27b7a750f1876632e84d594608d3d64ce974efc (patch) | |
tree | d7f973387b4ba1776cacbf06dcffef8a9428e9fe /server/lib | |
parent | 3396e6534592865f184ee2db32a75957c42cb887 (diff) | |
download | PeerTube-f27b7a750f1876632e84d594608d3d64ce974efc.tar.gz PeerTube-f27b7a750f1876632e84d594608d3d64ce974efc.tar.zst PeerTube-f27b7a750f1876632e84d594608d3d64ce974efc.zip |
Send views in a dedicated queue
Diffstat (limited to 'server/lib')
-rw-r--r-- | server/lib/activitypub/send/send-view.ts | 2 | ||||
-rw-r--r-- | server/lib/activitypub/send/shared/send-utils.ts | 19 | ||||
-rw-r--r-- | server/lib/job-queue/job-queue.ts | 3 |
3 files changed, 19 insertions, 5 deletions
diff --git a/server/lib/activitypub/send/send-view.ts b/server/lib/activitypub/send/send-view.ts index 25a20ec6d..bf3451603 100644 --- a/server/lib/activitypub/send/send-view.ts +++ b/server/lib/activitypub/send/send-view.ts | |||
@@ -26,7 +26,7 @@ async function sendView (options: { | |||
26 | return buildViewActivity({ url, byActor, video, audience, type }) | 26 | return buildViewActivity({ url, byActor, video, audience, type }) |
27 | } | 27 | } |
28 | 28 | ||
29 | return sendVideoRelatedActivity(activityBuilder, { byActor, video, transaction, contextType: 'View' }) | 29 | return sendVideoRelatedActivity(activityBuilder, { byActor, video, transaction, contextType: 'View', parallelizable: true }) |
30 | } | 30 | } |
31 | 31 | ||
32 | // --------------------------------------------------------------------------- | 32 | // --------------------------------------------------------------------------- |
diff --git a/server/lib/activitypub/send/shared/send-utils.ts b/server/lib/activitypub/send/shared/send-utils.ts index 72368c762..fcec63991 100644 --- a/server/lib/activitypub/send/shared/send-utils.ts +++ b/server/lib/activitypub/send/shared/send-utils.ts | |||
@@ -15,9 +15,10 @@ async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAud | |||
15 | byActor: MActorLight | 15 | byActor: MActorLight |
16 | video: MVideoImmutable | MVideoAccountLight | 16 | video: MVideoImmutable | MVideoAccountLight |
17 | contextType: ContextType | 17 | contextType: ContextType |
18 | parallelizable?: boolean | ||
18 | transaction?: Transaction | 19 | transaction?: Transaction |
19 | }) { | 20 | }) { |
20 | const { byActor, video, transaction, contextType } = options | 21 | const { byActor, video, transaction, contextType, parallelizable } = options |
21 | 22 | ||
22 | // Send to origin | 23 | // Send to origin |
23 | if (video.isOwned() === false) { | 24 | if (video.isOwned() === false) { |
@@ -38,6 +39,7 @@ async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAud | |||
38 | toFollowersOf: actorsInvolvedInVideo, | 39 | toFollowersOf: actorsInvolvedInVideo, |
39 | transaction, | 40 | transaction, |
40 | actorsException, | 41 | actorsException, |
42 | parallelizable, | ||
41 | contextType | 43 | contextType |
42 | }) | 44 | }) |
43 | } | 45 | } |
@@ -130,9 +132,10 @@ async function broadcastToFollowers (options: { | |||
130 | transaction: Transaction | 132 | transaction: Transaction |
131 | contextType: ContextType | 133 | contextType: ContextType |
132 | 134 | ||
135 | parallelizable?: boolean | ||
133 | actorsException?: MActorWithInboxes[] | 136 | actorsException?: MActorWithInboxes[] |
134 | }) { | 137 | }) { |
135 | const { data, byActor, toFollowersOf, transaction, contextType, actorsException = [] } = options | 138 | const { data, byActor, toFollowersOf, transaction, contextType, actorsException = [], parallelizable } = options |
136 | 139 | ||
137 | const uris = await computeFollowerUris(toFollowersOf, actorsException, transaction) | 140 | const uris = await computeFollowerUris(toFollowersOf, actorsException, transaction) |
138 | 141 | ||
@@ -141,6 +144,7 @@ async function broadcastToFollowers (options: { | |||
141 | uris, | 144 | uris, |
142 | data, | 145 | data, |
143 | byActor, | 146 | byActor, |
147 | parallelizable, | ||
144 | contextType | 148 | contextType |
145 | }) | 149 | }) |
146 | }) | 150 | }) |
@@ -173,8 +177,9 @@ function broadcastTo (options: { | |||
173 | data: any | 177 | data: any |
174 | byActor: MActorId | 178 | byActor: MActorId |
175 | contextType: ContextType | 179 | contextType: ContextType |
180 | parallelizable?: boolean // default to false | ||
176 | }) { | 181 | }) { |
177 | const { uris, data, byActor, contextType } = options | 182 | const { uris, data, byActor, contextType, parallelizable } = options |
178 | 183 | ||
179 | if (uris.length === 0) return undefined | 184 | if (uris.length === 0) return undefined |
180 | 185 | ||
@@ -200,7 +205,13 @@ function broadcastTo (options: { | |||
200 | contextType | 205 | contextType |
201 | } | 206 | } |
202 | 207 | ||
203 | JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload }) | 208 | JobQueue.Instance.createJob({ |
209 | type: parallelizable | ||
210 | ? 'activitypub-http-broadcast-parallel' | ||
211 | : 'activitypub-http-broadcast', | ||
212 | |||
213 | payload | ||
214 | }) | ||
204 | } | 215 | } |
205 | 216 | ||
206 | for (const unicastUri of unicastUris) { | 217 | for (const unicastUri of unicastUris) { |
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index d3776c3bf..f339e9135 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts | |||
@@ -43,6 +43,7 @@ import { processVideosViewsStats } from './handlers/video-views-stats' | |||
43 | 43 | ||
44 | type CreateJobArgument = | 44 | type CreateJobArgument = |
45 | { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | | 45 | { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | |
46 | { type: 'activitypub-http-broadcast-parallel', payload: ActivitypubHttpBroadcastPayload } | | ||
46 | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | | 47 | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | |
47 | { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | | 48 | { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | |
48 | { type: 'activitypub-http-cleaner', payload: {} } | | 49 | { type: 'activitypub-http-cleaner', payload: {} } | |
@@ -68,6 +69,7 @@ export type CreateJobOptions = { | |||
68 | 69 | ||
69 | const handlers: { [id in JobType]: (job: Job) => Promise<any> } = { | 70 | const handlers: { [id in JobType]: (job: Job) => Promise<any> } = { |
70 | 'activitypub-http-broadcast': processActivityPubHttpBroadcast, | 71 | 'activitypub-http-broadcast': processActivityPubHttpBroadcast, |
72 | 'activitypub-http-broadcast-parallel': processActivityPubHttpBroadcast, | ||
71 | 'activitypub-http-unicast': processActivityPubHttpUnicast, | 73 | 'activitypub-http-unicast': processActivityPubHttpUnicast, |
72 | 'activitypub-http-fetcher': processActivityPubHttpFetcher, | 74 | 'activitypub-http-fetcher': processActivityPubHttpFetcher, |
73 | 'activitypub-cleaner': processActivityPubCleaner, | 75 | 'activitypub-cleaner': processActivityPubCleaner, |
@@ -89,6 +91,7 @@ const handlers: { [id in JobType]: (job: Job) => Promise<any> } = { | |||
89 | const jobTypes: JobType[] = [ | 91 | const jobTypes: JobType[] = [ |
90 | 'activitypub-follow', | 92 | 'activitypub-follow', |
91 | 'activitypub-http-broadcast', | 93 | 'activitypub-http-broadcast', |
94 | 'activitypub-http-broadcast-parallel', | ||
92 | 'activitypub-http-fetcher', | 95 | 'activitypub-http-fetcher', |
93 | 'activitypub-http-unicast', | 96 | 'activitypub-http-unicast', |
94 | 'activitypub-cleaner', | 97 | 'activitypub-cleaner', |