limepotato/jormungandr-bite
1
0
Fork 0
mirror of https://iceshrimp.dev/limepotato/jormungandr-bite.git synced 2024-11-22 01:47:39 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

perf: 各ストリーミング接続ごとにポーリングしないように

Browse source
This commit is contained in:
syuilo 2021-03-21 15:14:03 +09:00
parent df46cbe878
commit 84e7dee3c8
11 changed files with 83 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/server/api/endpoints/channels/follow.ts
View file

@ -4,6 +4,7 @@ import define from '../../define';
import { ApiError } from '../../error';
import { Channels, ChannelFollowings } from '../../../../models';
import { genId } from '../../../../misc/gen-id';
import { publishUserEvent } from '../../../../services/stream';
export const meta = {
tags: ['channels'],
@ -42,4 +43,6 @@ export default define(meta, async (ps, user) => {
followerId: user.id,
followeeId: channel.id,
});
publishUserEvent(user.id, 'followChannel', channel);
});

3
src/server/api/endpoints/channels/unfollow.ts
View file

@ -3,6 +3,7 @@ import { ID } from '../../../../misc/cafy-id';
import define from '../../define';
import { ApiError } from '../../error';
import { Channels, ChannelFollowings } from '../../../../models';
import { publishUserEvent } from '../../../../services/stream';
export const meta = {
tags: ['channels'],
@ -39,4 +40,6 @@ export default define(meta, async (ps, user) => {
followerId: user.id,
followeeId: channel.id,
});
publishUserEvent(user.id, 'unfollowChannel', channel);
});

3
src/server/api/endpoints/i/update.ts
View file

@ -1,6 +1,6 @@
import $ from 'cafy';
import { ID } from '../../../../misc/cafy-id';
import { publishMainStream } from '../../../../services/stream';
import { publishMainStream, publishUserEvent } from '../../../../services/stream';
import acceptAllFollowRequests from '../../../../services/following/requests/accept-all';
import { publishToFollowers } from '../../../../services/i/update';
import define from '../../define';
@ -317,6 +317,7 @@ export default define(meta, async (ps, user, token) => {
// Publish meUpdated event
publishMainStream(user.id, 'meUpdated', iObj);
publishUserEvent(user.id, 'updateUserProfile', await UserProfiles.findOne(user.id));
// 鍵垢を解除したとき、溜まっていたフォローリクエストがあるならすべて承認
if (user.isLocked && ps.isLocked === false) {

3
src/server/api/endpoints/mute/create.ts
View file

@ -6,6 +6,7 @@ import { getUser } from '../../common/getters';
import { genId } from '../../../../misc/gen-id';
import { Mutings, NoteWatchings } from '../../../../models';
import { Muting } from '../../../../models/entities/muting';
import { publishUserEvent } from '../../../../services/stream';
export const meta = {
desc: {
@ -82,6 +83,8 @@ export default define(meta, async (ps, user) => {
muteeId: mutee.id,
} as Muting);
publishUserEvent(user.id, 'mute', mutee);
NoteWatchings.delete({
userId: muter.id,
noteUserId: mutee.id

3
src/server/api/endpoints/mute/delete.ts
View file

@ -4,6 +4,7 @@ import define from '../../define';
import { ApiError } from '../../error';
import { getUser } from '../../common/getters';
import { Mutings } from '../../../../models';
import { publishUserEvent } from '../../../../services/stream';
export const meta = {
desc: {
@ -76,4 +77,6 @@ export default define(meta, async (ps, user) => {
await Mutings.delete({
id: exist.id
});
publishUserEvent(user.id, 'unmute', mutee);
});

57
src/server/api/stream/index.ts
View file

@ -30,10 +30,6 @@ export default class Connection {
public subscriber: EventEmitter;
private channels: Channel[] = [];
private subscribingNotes: any = {};
private followingClock: ReturnType<typeof setInterval>;
private mutingClock: ReturnType<typeof setInterval>;
private followingChannelsClock: ReturnType<typeof setInterval>;
private userProfileClock: ReturnType<typeof setInterval>;
constructor(
wsConnection: websocket.connection,
@ -52,19 +48,51 @@ export default class Connection {
this.onBroadcastMessage(type, body);
});
// TODO: reidsでイベントをもらうようにし、ポーリングはしないようにする
if (this.user) {
this.updateFollowing();
this.followingClock = setInterval(this.updateFollowing, 5000);
this.updateMuting();
this.mutingClock = setInterval(this.updateMuting, 5000);
this.updateFollowingChannels();
this.followingChannelsClock = setInterval(this.updateFollowingChannels, 5000);
this.updateUserProfile();
this.userProfileClock = setInterval(this.updateUserProfile, 5000);
this.subscriber.on(`user:${this.user.id}`, ({ type, body }) => {
this.onUserEvent(type, body);
});
}
}
@autobind
private onUserEvent(type: string, body: any) {
switch (type) {
case 'follow':
this.following.add(body.id);
break;
case 'unfollow':
this.following.delete(body.id);
break;
case 'mute':
this.muting.add(body.id);
break;
case 'unmute':
this.muting.delete(body.id);
break;
case 'followChannel':
this.followingChannels.add(body.id);
break;
case 'unfollowChannel':
this.followingChannels.delete(body.id);
break;
case 'updateUserProfile':
this.userProfile = body;
break;
default:
break;
}
}
@ -354,10 +382,5 @@ export default class Connection {
for (const c of this.channels.filter(c => c.dispose)) {
if (c.dispose) c.dispose();
}
if (this.followingClock) clearInterval(this.followingClock);
if (this.mutingClock) clearInterval(this.mutingClock);
if (this.followingChannelsClock) clearInterval(this.followingChannelsClock);
if (this.userProfileClock) clearInterval(this.userProfileClock);
}
}

12
src/services/blocking/create.ts
View file

@ -1,4 +1,4 @@
import { publishMainStream } from '../stream';
import { publishMainStream, publishUserEvent } from '../stream';
import { renderActivity } from '../../remote/activitypub/renderer';
import renderFollow from '../../remote/activitypub/renderer/follow';
import renderUndo from '../../remote/activitypub/renderer/undo';
@ -55,7 +55,10 @@ async function cancelRequest(follower: User, followee: User) {
if (Users.isLocalUser(follower)) {
Users.pack(followee, follower, {
detail: true
}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
}).then(packed => {
publishUserEvent(follower.id, 'unfollow', packed);
publishMainStream(follower.id, 'unfollow', packed);
});
}
// リモートにフォローリクエストをしていたらUndoFollow送信
@ -97,7 +100,10 @@ async function unFollow(follower: User, followee: User) {
if (Users.isLocalUser(follower)) {
Users.pack(followee, follower, {
detail: true
}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
}).then(packed => {
publishUserEvent(follower.id, 'unfollow', packed);
publishMainStream(follower.id, 'unfollow', packed);
});
}
// リモートにフォローをしていたらUndoFollow送信

7
src/services/following/create.ts
View file

@ -1,4 +1,4 @@
import { publishMainStream } from '../stream';
import { publishMainStream, publishUserEvent } from '../stream';
import { renderActivity } from '../../remote/activitypub/renderer';
import renderFollow from '../../remote/activitypub/renderer/follow';
import renderAccept from '../../remote/activitypub/renderer/accept';
@ -88,7 +88,10 @@ export async function insertFollowingDoc(followee: User, follower: User) {
if (Users.isLocalUser(follower)) {
Users.pack(followee, follower, {
detail: true
}).then(packed => publishMainStream(follower.id, 'follow', packed));
}).then(packed => {
publishUserEvent(follower.id, 'follow', packed);
publishMainStream(follower.id, 'follow', packed);
});
}
// Publish followed event

7
src/services/following/delete.ts
View file

@ -1,4 +1,4 @@
import { publishMainStream } from '../stream';
import { publishMainStream, publishUserEvent } from '../stream';
import { renderActivity } from '../../remote/activitypub/renderer';
import renderFollow from '../../remote/activitypub/renderer/follow';
import renderUndo from '../../remote/activitypub/renderer/undo';
@ -30,7 +30,10 @@ export default async function(follower: User, followee: User, silent = false) {
if (!silent && Users.isLocalUser(follower)) {
Users.pack(followee, follower, {
detail: true
}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
}).then(packed => {
publishUserEvent(follower.id, 'unfollow', packed);
publishMainStream(follower.id, 'unfollow', packed);
});
}
if (Users.isLocalUser(follower) && Users.isRemoteUser(followee)) {

7
src/services/following/requests/reject.ts
View file

@ -2,7 +2,7 @@ import { renderActivity } from '../../../remote/activitypub/renderer';
import renderFollow from '../../../remote/activitypub/renderer/follow';
import renderReject from '../../../remote/activitypub/renderer/reject';
import { deliver } from '../../../queue';
import { publishMainStream } from '../../stream';
import { publishMainStream, publishUserEvent } from '../../stream';
import { User, ILocalUser } from '../../../models/entities/user';
import { Users, FollowRequests, Followings } from '../../../models';
import { decrementFollowing } from '../delete';
@ -39,5 +39,8 @@ export default async function(followee: User, follower: User) {
Users.pack(followee, follower, {
detail: true
}).then(packed => publishMainStream(follower.id, 'unfollow', packed));
}).then(packed => {
publishUserEvent(follower.id, 'unfollow', packed);
publishMainStream(follower.id, 'unfollow', packed);
});
}

5
src/services/stream.ts
View file

@ -20,6 +20,10 @@ class Publisher {
}));
}
public publishUserEvent = (userId: User['id'], type: string, value?: any): void => {
this.publish(`user:${userId}`, type, typeof value === 'undefined' ? null : value);
}
public publishBroadcastStream = (type: string, value?: any): void => {
this.publish('broadcast', type, typeof value === 'undefined' ? null : value);
}
@ -84,6 +88,7 @@ const publisher = new Publisher();
export default publisher;
export const publishUserEvent = publisher.publishUserEvent;
export const publishBroadcastStream = publisher.publishBroadcastStream;
export const publishMainStream = publisher.publishMainStream;
export const publishDriveStream = publisher.publishDriveStream;