diff options
author | Chocobozzz <me@florianbigard.com> | 2018-01-25 15:05:18 +0100 |
---|---|---|
committer | Chocobozzz <me@florianbigard.com> | 2018-01-25 18:41:17 +0100 |
commit | 94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4 (patch) | |
tree | 32a9148e0e4567f0c4ffae0412cbed20b84e8873 /server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts | |
parent | d765fafc3faf0db9818eb1a07161df1cb1bc0efa (diff) | |
download | PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.gz PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.tar.zst PeerTube-94a5ff8a4a75d75bb9df542a39ce8769e7a7e6a4.zip |
Move job queue to redis
We'll use it as cache in the future.
/!\ You'll loose your old jobs (pending jobs too) so upgrade only when
you don't have pending job anymore.
Diffstat (limited to 'server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts')
-rw-r--r-- | server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts | 68 |
1 files changed, 0 insertions, 68 deletions
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts deleted file mode 100644 index a7b5aabd0..000000000 --- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts +++ /dev/null | |||
@@ -1,68 +0,0 @@ | |||
1 | import { logger } from '../../../helpers/logger' | ||
2 | import { doRequest } from '../../../helpers/requests' | ||
3 | import { ACTIVITY_PUB } from '../../../initializers' | ||
4 | import { processActivities } from '../../activitypub/process' | ||
5 | import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler' | ||
6 | |||
7 | async function process (payload: ActivityPubHttpPayload, jobId: number) { | ||
8 | logger.info('Processing ActivityPub fetcher in job %d.', jobId) | ||
9 | |||
10 | const options = { | ||
11 | method: 'GET', | ||
12 | uri: '', | ||
13 | json: true, | ||
14 | activityPub: true | ||
15 | } | ||
16 | |||
17 | for (const uri of payload.uris) { | ||
18 | options.uri = uri | ||
19 | logger.info('Fetching ActivityPub data on %s.', uri) | ||
20 | |||
21 | const response = await doRequest(options) | ||
22 | const firstBody = response.body | ||
23 | |||
24 | if (firstBody.first && Array.isArray(firstBody.first.orderedItems)) { | ||
25 | const activities = firstBody.first.orderedItems | ||
26 | |||
27 | logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri) | ||
28 | |||
29 | await processActivities(activities) | ||
30 | } | ||
31 | |||
32 | let limit = ACTIVITY_PUB.FETCH_PAGE_LIMIT | ||
33 | let i = 0 | ||
34 | let nextLink = firstBody.first.next | ||
35 | while (nextLink && i < limit) { | ||
36 | options.uri = nextLink | ||
37 | |||
38 | const { body } = await doRequest(options) | ||
39 | nextLink = body.next | ||
40 | i++ | ||
41 | |||
42 | if (Array.isArray(body.orderedItems)) { | ||
43 | const activities = body.orderedItems | ||
44 | logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri) | ||
45 | |||
46 | await processActivities(activities) | ||
47 | } | ||
48 | } | ||
49 | } | ||
50 | } | ||
51 | |||
52 | function onError (err: Error, jobId: number) { | ||
53 | logger.error('Error when fetcher ActivityPub request in job %d.', jobId, err) | ||
54 | return Promise.resolve() | ||
55 | } | ||
56 | |||
57 | function onSuccess (jobId: number) { | ||
58 | logger.info('Job %d is a success.', jobId) | ||
59 | return Promise.resolve() | ||
60 | } | ||
61 | |||
62 | // --------------------------------------------------------------------------- | ||
63 | |||
64 | export { | ||
65 | process, | ||
66 | onError, | ||
67 | onSuccess | ||
68 | } | ||