Typed stream (#13)

* Update streaming.ts

* Update streaming.ts

* wip
This commit is contained in:
syuilo 2021-05-17 19:50:31 +09:00 committed by GitHub
parent d7d02cd2bc
commit 99276028ae
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 174 additions and 10 deletions

View file

@ -1,5 +1,7 @@
export type ID = string; export type ID = string;
type TODO = Record<string, any>;
export type User = { export type User = {
id: ID; id: ID;
username: string; username: string;
@ -14,6 +16,17 @@ export type User = {
}[]; }[];
}; };
export type MeDetailed = User & {
avatarId: DriveFile['id'];
bannerId: DriveFile['id'];
autoAcceptFollowed: boolean;
noCrawle: boolean;
isExplorable: boolean;
hideOnlineStatus: boolean;
mutedWords: string[][];
[other: string]: any;
};
export type DriveFile = { export type DriveFile = {
id: ID; id: ID;
createdAt: string; createdAt: string;
@ -59,6 +72,74 @@ export type Note = {
}[]; }[];
}; };
export type Notification = {
id: ID;
createdAt: string;
isRead: boolean;
} & ({
type: 'reaction';
reaction: string;
user: User;
userId: User['id'];
note: Note;
} | {
type: 'reply';
user: User;
userId: User['id'];
note: Note;
} | {
type: 'renote';
user: User;
userId: User['id'];
note: Note;
} | {
type: 'quote';
user: User;
userId: User['id'];
note: Note;
} | {
type: 'mention';
user: User;
userId: User['id'];
note: Note;
} | {
type: 'pollVote';
user: User;
userId: User['id'];
note: Note;
} | {
type: 'follow';
user: User;
userId: User['id'];
} | {
type: 'followRequestAccepted';
user: User;
userId: User['id'];
} | {
type: 'receiveFollowRequest';
user: User;
userId: User['id'];
} | {
type: 'groupInvited'; // TODO
} | {
type: 'app';
body: string;
icon: string;
});
export type MessagingMessage = {
id: ID;
createdAt: string;
file: DriveFile | null;
fileId: DriveFile['id'] | null;
isRead: boolean;
reads: User['id'][];
text: string | null;
user: User;
userId: User['id'];
groupId: string; // TODO
};
export type InstanceMetadata = { export type InstanceMetadata = {
emojis: { emojis: {
category: string; category: string;
@ -119,5 +200,13 @@ export type Page = {
isLiked?: boolean; isLiked?: boolean;
}; };
export type PageEvent = {
pageId: Page['id'];
event: string;
var: any;
userId: User['id'];
user: User;
};
export type UserSorting = '+follower' | '-follower' | '+createdAt' | '-createdAt' | '+updatedAt' | '-updatedAt'; export type UserSorting = '+follower' | '-follower' | '+createdAt' | '-createdAt' | '+updatedAt' | '-updatedAt';
export type OriginType = 'combined' | 'local' | 'remote'; export type OriginType = 'combined' | 'local' | 'remote';

View file

@ -3,6 +3,7 @@ import { EventEmitter } from 'eventemitter3';
import ReconnectingWebsocket from 'reconnecting-websocket'; import ReconnectingWebsocket from 'reconnecting-websocket';
import { stringify } from 'querystring'; import { stringify } from 'querystring';
import { markRaw } from '@vue/reactivity'; import { markRaw } from '@vue/reactivity';
import { MeDetailed, MessagingMessage, Note, Notification, PageEvent, User } from './entities';
function urlQuery(obj: {}): string { function urlQuery(obj: {}): string {
return stringify(Object.entries(obj) return stringify(Object.entries(obj)
@ -10,10 +11,84 @@ function urlQuery(obj: {}): string {
.reduce((a, [k, v]) => (a[k] = v, a), {} as Record<string, any>)); .reduce((a, [k, v]) => (a[k] = v, a), {} as Record<string, any>));
} }
type FIXME = any;
type ChannelDef = {
main: {
events: {
notification: (payload: Notification) => void;
mention: (payload: Note) => void;
reply: (payload: Note) => void;
renote: (payload: Note) => void;
follow: (payload: User) => void; // 自分が他人をフォローしたとき
followed: (payload: User) => void; // 他人が自分をフォローしたとき
unfollow: (payload: User) => void; // 自分が他人をフォロー解除したとき
meUpdated: (payload: MeDetailed) => void;
pageEvent: (payload: PageEvent) => void;
};
};
homeTimeline: {
events: {
note: (payload: Note) => void;
};
};
localTimeline: {
events: {
note: (payload: Note) => void;
};
};
hybridTimeline: {
events: {
note: (payload: Note) => void;
};
};
globalTimeline: {
events: {
note: (payload: Note) => void;
};
};
messaging: {
events: {
message: (payload: MessagingMessage) => void;
deleted: (payload: MessagingMessage['id']) => void;
read: (payload: MessagingMessage['id'][]) => void;
typing: (payload: User['id']) => void;
};
};
};
type NoteUpdatedEvent = {
id: Note['id'];
type: 'reacted';
body: {
reaction: string;
userId: User['id'];
};
} | {
id: Note['id'];
type: 'deleted';
body: {
deletedAt: string;
};
} | {
id: Note['id'];
type: 'pollVoted';
body: {
choice: number;
userId: User['id'];
};
};
type StreamEvents = {
_connected_: void;
_disconnected_: void;
noteUpdated: (payload: NoteUpdatedEvent) => void;
};
/** /**
* Misskey stream connection * Misskey stream connection
*/ */
export default class Stream extends EventEmitter { export default class Stream extends EventEmitter<StreamEvents> {
private stream: ReconnectingWebsocket; private stream: ReconnectingWebsocket;
public state: 'initializing' | 'reconnecting' | 'connected' = 'initializing'; public state: 'initializing' | 'reconnecting' | 'connected' = 'initializing';
private sharedConnectionPools: Pool[] = []; private sharedConnectionPools: Pool[] = [];
@ -38,7 +113,7 @@ export default class Stream extends EventEmitter {
} }
@autobind @autobind
public useSharedConnection(channel: string, name?: string): SharedConnection { public useSharedConnection<C extends keyof ChannelDef>(channel: C, name?: string): SharedConnection<ChannelDef[C]['events']> {
let pool = this.sharedConnectionPools.find(p => p.channel === channel); let pool = this.sharedConnectionPools.find(p => p.channel === channel);
if (pool == null) { if (pool == null) {
@ -62,7 +137,7 @@ export default class Stream extends EventEmitter {
} }
@autobind @autobind
public connectToChannel(channel: string, params?: any): NonSharedConnection { public connectToChannel<C extends keyof ChannelDef>(channel: C, params?: any): NonSharedConnection<ChannelDef[C]['events']> {
const connection = markRaw(new NonSharedConnection(this, channel, params)); const connection = markRaw(new NonSharedConnection(this, channel, params));
this.nonSharedConnections.push(connection); this.nonSharedConnections.push(connection);
return connection; return connection;
@ -227,7 +302,7 @@ class Pool {
} }
} }
abstract class Connection extends EventEmitter { abstract class Connection<Events extends Record<string, any> = any> extends EventEmitter<Events> {
public channel: string; public channel: string;
protected stream: Stream; protected stream: Stream;
public abstract id: string; public abstract id: string;
@ -261,7 +336,7 @@ abstract class Connection extends EventEmitter {
public abstract dispose(): void; public abstract dispose(): void;
} }
class SharedConnection extends Connection { class SharedConnection<Events = any> extends Connection<Events> {
private pool: Pool; private pool: Pool;
public get id(): string { public get id(): string {
@ -288,7 +363,7 @@ class SharedConnection extends Connection {
} }
} }
class NonSharedConnection extends Connection { class NonSharedConnection<Events = any> extends Connection<Events> {
public id: string; public id: string;
protected params: any; protected params: any;

View file

@ -7,7 +7,7 @@ describe('Streaming', () => {
const stream = new Stream('https://misskey.test', { token: 'TOKEN' }); const stream = new Stream('https://misskey.test', { token: 'TOKEN' });
const mainChannelReceived: any[] = []; const mainChannelReceived: any[] = [];
const main = stream.useSharedConnection('main'); const main = stream.useSharedConnection('main');
main.on('foo', payload => { main.on('meUpdated', payload => {
mainChannelReceived.push(payload); mainChannelReceived.push(payload);
}); });
await server.connected; await server.connected;
@ -21,15 +21,15 @@ describe('Streaming', () => {
type: 'channel', type: 'channel',
body: { body: {
id: mainChannelId, id: mainChannelId,
type: 'foo', type: 'meUpdated',
body: { body: {
bar: 'buzz' id: 'foo'
} }
} }
})); }));
expect(mainChannelReceived[0]).toEqual({ expect(mainChannelReceived[0]).toEqual({
bar: 'buzz' id: 'foo'
}); });
}); });
}); });