diff options
author | Chocobozzz <florian.bigard@gmail.com> | 2017-11-17 11:35:10 +0100 |
---|---|---|
committer | Chocobozzz <florian.bigard@gmail.com> | 2017-11-27 19:40:52 +0100 |
commit | afffe98839db7ccbfa9fb8b7d1413b97900fdc73 (patch) | |
tree | f0b3cbe58f73e81a5ba4bb31dabd9691994cf2ca /server/lib/jobs | |
parent | 1b3989b0961d22a5a45ad16239e3c3f58f66180c (diff) | |
download | PeerTube-afffe98839db7ccbfa9fb8b7d1413b97900fdc73.tar.gz PeerTube-afffe98839db7ccbfa9fb8b7d1413b97900fdc73.tar.zst PeerTube-afffe98839db7ccbfa9fb8b7d1413b97900fdc73.zip |
Speed up activity pub http requests
Diffstat (limited to 'server/lib/jobs')
9 files changed, 108 insertions, 94 deletions
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-broadcast-handler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-broadcast-handler.ts new file mode 100644 index 000000000..111fc88a4 --- /dev/null +++ b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-broadcast-handler.ts | |||
@@ -0,0 +1,43 @@ | |||
1 | import { logger } from '../../../helpers' | ||
2 | import { buildSignedActivity } from '../../../helpers/activitypub' | ||
3 | import { doRequest } from '../../../helpers/requests' | ||
4 | import { database as db } from '../../../initializers' | ||
5 | import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler' | ||
6 | |||
7 | async function process (payload: ActivityPubHttpPayload, jobId: number) { | ||
8 | logger.info('Processing ActivityPub broadcast in job %d.', jobId) | ||
9 | |||
10 | const accountSignature = await db.Account.load(payload.signatureAccountId) | ||
11 | if (!accountSignature) throw new Error('Unknown signature account id.') | ||
12 | |||
13 | const signedBody = await buildSignedActivity(accountSignature, payload.body) | ||
14 | |||
15 | const options = { | ||
16 | method: 'POST', | ||
17 | uri: '', | ||
18 | json: signedBody | ||
19 | } | ||
20 | |||
21 | for (const uri of payload.uris) { | ||
22 | options.uri = uri | ||
23 | await doRequest(options) | ||
24 | } | ||
25 | } | ||
26 | |||
27 | function onError (err: Error, jobId: number) { | ||
28 | logger.error('Error when broadcasting ActivityPub request in job %d.', jobId, err) | ||
29 | return Promise.resolve() | ||
30 | } | ||
31 | |||
32 | function onSuccess (jobId: number) { | ||
33 | logger.info('Job %d is a success.', jobId) | ||
34 | return Promise.resolve() | ||
35 | } | ||
36 | |||
37 | // --------------------------------------------------------------------------- | ||
38 | |||
39 | export { | ||
40 | process, | ||
41 | onError, | ||
42 | onSuccess | ||
43 | } | ||
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts new file mode 100644 index 000000000..e4f6c94a5 --- /dev/null +++ b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts | |||
@@ -0,0 +1,23 @@ | |||
1 | import { JobScheduler, JobHandler } from '../job-scheduler' | ||
2 | |||
3 | import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler' | ||
4 | import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler' | ||
5 | import { JobCategory } from '../../../../shared' | ||
6 | |||
7 | type ActivityPubHttpPayload = { | ||
8 | uris: string[] | ||
9 | signatureAccountId: number | ||
10 | body: any | ||
11 | } | ||
12 | const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = { | ||
13 | activitypubHttpBroadcastHandler, | ||
14 | activitypubHttpUnicastHandler | ||
15 | } | ||
16 | const jobCategory: JobCategory = 'activitypub-http' | ||
17 | |||
18 | const activitypubHttpJobScheduler = new JobScheduler(jobCategory, jobHandlers) | ||
19 | |||
20 | export { | ||
21 | ActivityPubHttpPayload, | ||
22 | activitypubHttpJobScheduler | ||
23 | } | ||
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts new file mode 100644 index 000000000..8d3b755ad --- /dev/null +++ b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts | |||
@@ -0,0 +1,40 @@ | |||
1 | import { logger } from '../../../helpers' | ||
2 | import { doRequest } from '../../../helpers/requests' | ||
3 | import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler' | ||
4 | import { database as db } from '../../../initializers/database' | ||
5 | import { buildSignedActivity } from '../../../helpers/activitypub' | ||
6 | |||
7 | async function process (payload: ActivityPubHttpPayload, jobId: number) { | ||
8 | logger.info('Processing ActivityPub unicast in job %d.', jobId) | ||
9 | |||
10 | const accountSignature = await db.Account.load(payload.signatureAccountId) | ||
11 | if (!accountSignature) throw new Error('Unknown signature account id.') | ||
12 | |||
13 | const signedBody = await buildSignedActivity(accountSignature, payload.body) | ||
14 | const uri = payload.uris[0] | ||
15 | const options = { | ||
16 | method: 'POST', | ||
17 | uri, | ||
18 | json: signedBody | ||
19 | } | ||
20 | |||
21 | await doRequest(options) | ||
22 | } | ||
23 | |||
24 | function onError (err: Error, jobId: number) { | ||
25 | logger.error('Error when sending ActivityPub request in job %d.', jobId, err) | ||
26 | return Promise.resolve() | ||
27 | } | ||
28 | |||
29 | function onSuccess (jobId: number) { | ||
30 | logger.info('Job %d is a success.', jobId) | ||
31 | return Promise.resolve() | ||
32 | } | ||
33 | |||
34 | // --------------------------------------------------------------------------- | ||
35 | |||
36 | export { | ||
37 | process, | ||
38 | onError, | ||
39 | onSuccess | ||
40 | } | ||
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/index.ts b/server/lib/jobs/activitypub-http-job-scheduler/index.ts new file mode 100644 index 000000000..ad8f527b4 --- /dev/null +++ b/server/lib/jobs/activitypub-http-job-scheduler/index.ts | |||
@@ -0,0 +1 @@ | |||
export * from './activitypub-http-job-scheduler' | |||
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts deleted file mode 100644 index ccb008e4d..000000000 --- a/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts +++ /dev/null | |||
@@ -1,36 +0,0 @@ | |||
1 | import { logger } from '../../../helpers' | ||
2 | import { doRequest } from '../../../helpers/requests' | ||
3 | import { HTTPRequestPayload } from './http-request-job-scheduler' | ||
4 | |||
5 | async function process (payload: HTTPRequestPayload, jobId: number) { | ||
6 | logger.info('Processing broadcast in job %d.', jobId) | ||
7 | |||
8 | const options = { | ||
9 | method: 'POST', | ||
10 | uri: '', | ||
11 | json: payload.body | ||
12 | } | ||
13 | |||
14 | for (const uri of payload.uris) { | ||
15 | options.uri = uri | ||
16 | await doRequest(options) | ||
17 | } | ||
18 | } | ||
19 | |||
20 | function onError (err: Error, jobId: number) { | ||
21 | logger.error('Error when broadcasting request in job %d.', jobId, err) | ||
22 | return Promise.resolve() | ||
23 | } | ||
24 | |||
25 | function onSuccess (jobId: number) { | ||
26 | logger.info('Job %d is a success.', jobId) | ||
27 | return Promise.resolve() | ||
28 | } | ||
29 | |||
30 | // --------------------------------------------------------------------------- | ||
31 | |||
32 | export { | ||
33 | process, | ||
34 | onError, | ||
35 | onSuccess | ||
36 | } | ||
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts deleted file mode 100644 index ad3349866..000000000 --- a/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts +++ /dev/null | |||
@@ -1,22 +0,0 @@ | |||
1 | import { JobScheduler, JobHandler } from '../job-scheduler' | ||
2 | |||
3 | import * as httpRequestBroadcastHandler from './http-request-broadcast-handler' | ||
4 | import * as httpRequestUnicastHandler from './http-request-unicast-handler' | ||
5 | import { JobCategory } from '../../../../shared' | ||
6 | |||
7 | type HTTPRequestPayload = { | ||
8 | uris: string[] | ||
9 | body: any | ||
10 | } | ||
11 | const jobHandlers: { [ handlerName: string ]: JobHandler<HTTPRequestPayload, void> } = { | ||
12 | httpRequestBroadcastHandler, | ||
13 | httpRequestUnicastHandler | ||
14 | } | ||
15 | const jobCategory: JobCategory = 'http-request' | ||
16 | |||
17 | const httpRequestJobScheduler = new JobScheduler(jobCategory, jobHandlers) | ||
18 | |||
19 | export { | ||
20 | HTTPRequestPayload, | ||
21 | httpRequestJobScheduler | ||
22 | } | ||
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts deleted file mode 100644 index 9e4e73891..000000000 --- a/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts +++ /dev/null | |||
@@ -1,34 +0,0 @@ | |||
1 | import { logger } from '../../../helpers' | ||
2 | import { doRequest } from '../../../helpers/requests' | ||
3 | import { HTTPRequestPayload } from './http-request-job-scheduler' | ||
4 | |||
5 | async function process (payload: HTTPRequestPayload, jobId: number) { | ||
6 | logger.info('Processing unicast in job %d.', jobId) | ||
7 | |||
8 | const uri = payload.uris[0] | ||
9 | const options = { | ||
10 | method: 'POST', | ||
11 | uri, | ||
12 | json: payload.body | ||
13 | } | ||
14 | |||
15 | await doRequest(options) | ||
16 | } | ||
17 | |||
18 | function onError (err: Error, jobId: number) { | ||
19 | logger.error('Error when sending request in job %d.', jobId, err) | ||
20 | return Promise.resolve() | ||
21 | } | ||
22 | |||
23 | function onSuccess (jobId: number) { | ||
24 | logger.info('Job %d is a success.', jobId) | ||
25 | return Promise.resolve() | ||
26 | } | ||
27 | |||
28 | // --------------------------------------------------------------------------- | ||
29 | |||
30 | export { | ||
31 | process, | ||
32 | onError, | ||
33 | onSuccess | ||
34 | } | ||
diff --git a/server/lib/jobs/http-request-job-scheduler/index.ts b/server/lib/jobs/http-request-job-scheduler/index.ts deleted file mode 100644 index 4d2573296..000000000 --- a/server/lib/jobs/http-request-job-scheduler/index.ts +++ /dev/null | |||
@@ -1 +0,0 @@ | |||
1 | export * from './http-request-job-scheduler' | ||
diff --git a/server/lib/jobs/index.ts b/server/lib/jobs/index.ts index a92743707..394264ec1 100644 --- a/server/lib/jobs/index.ts +++ b/server/lib/jobs/index.ts | |||
@@ -1,2 +1,2 @@ | |||
1 | export * from './http-request-job-scheduler' | 1 | export * from './activitypub-http-job-scheduler' |
2 | export * from './transcoding-job-scheduler' | 2 | export * from './transcoding-job-scheduler' |