10.9.2

package.json

@@ -1,8 +1,8 @@
 {
 	"name": "misskey",
 	"author": "syuilo <i@syuilo.com>",
-	"version": "10.8.0",
-	"clientVersion": "1.0.10417",
+	"version": "10.9.2",
+	"clientVersion": "1.0.10448",
 	"codename": "nighthike",
 	"main": "./built/index.js",
 	"private": true,
@@ -169,6 +169,7 @@
 		"parse5": "5.1.0",
 		"portscanner": "2.2.0",
 		"progress-bar-webpack-plugin": "1.11.0",
+		"promise-limit": "2.7.0",
 		"promise-sequential": "1.1.1",
 		"pug": "2.0.3",
 		"punycode": "2.1.1",

src/client/app/common/scripts/stream.ts

@@ -10,7 +10,6 @@ import MiOS from '../../mios';
 export default class Stream extends EventEmitter {
 	private stream: ReconnectingWebsocket;
 	private state: string;
-	private buffer: any[];
 	private sharedConnectionPools: Pool[] = [];
 	private sharedConnections: SharedConnection[] = [];
 	private nonSharedConnections: NonSharedConnection[] = [];
@@ -19,7 +18,6 @@ export default class Stream extends EventEmitter {
 		super();
 
 		this.state = 'initializing';
-		this.buffer = [];
 
 		const user = os.store.state.i;
 
@@ -48,6 +46,11 @@ export default class Stream extends EventEmitter {
 		this.sharedConnections = this.sharedConnections.filter(c => c !== connection);
 	}
 
+	@autobind
+	public removeSharedConnectionPool(pool: Pool) {
+		this.sharedConnectionPools = this.sharedConnectionPools.filter(p => p !== pool);
+	}
+
 	@autobind
 	public connectToChannel(channel: string, params?: any): NonSharedConnection {
 		const connection = new NonSharedConnection(this, channel, params);
@@ -70,13 +73,6 @@ export default class Stream extends EventEmitter {
 		this.state = 'connected';
 		this.emit('_connected_');
 
-		// バッファーを処理
-		const _buffer = [].concat(this.buffer); // Shallow copy
-		this.buffer = []; // Clear buffer
-		_buffer.forEach(data => {
-			this.send(data); // Resend each buffered messages
-		});
-
 		// チャンネル再接続
 		if (isReconnect) {
 			this.sharedConnectionPools.forEach(p => {
@@ -93,8 +89,10 @@ export default class Stream extends EventEmitter {
 	 */
 	@autobind
 	private onClose() {
-		this.state = 'reconnecting';
-		this.emit('_disconnected_');
+		if (this.state == 'connected') {
+			this.state = 'reconnecting';
+			this.emit('_disconnected_');
+		}
 	}
 
 	/**
@@ -133,12 +131,6 @@ export default class Stream extends EventEmitter {
 			body: payload
 		};
 
-		// まだ接続が確立されていなかったらバッファリングして次に接続した時に送信する
-		if (this.state != 'connected') {
-			this.buffer.push(data);
-			return;
-		}
-
 		this.stream.send(JSON.stringify(data));
 	}
 
@@ -156,7 +148,7 @@ class Pool {
 	public channel: string;
 	public id: string;
 	protected stream: Stream;
-	private users = 0;
+	public users = 0;
 	private disposeTimerId: any;
 	private isConnected = false;
 
@@ -165,6 +157,13 @@ class Pool {
 		this.stream = stream;
 
 		this.id = Math.random().toString();
+
+		this.stream.on('_disconnected_', this.onStreamDisconnected);
+	}
+
+	@autobind
+	private onStreamDisconnected() {
+		this.isConnected = false;
 	}
 
 	@autobind
@@ -198,6 +197,7 @@ class Pool {
 
 	@autobind
 	public connect() {
+		if (this.isConnected) return;
 		this.isConnected = true;
 		this.stream.send('connect', {
 			channel: this.channel,
@@ -207,9 +207,9 @@ class Pool {
 
 	@autobind
 	private disconnect() {
-		this.isConnected = false;
-		this.disposeTimerId = null;
+		this.stream.off('_disconnected_', this.onStreamDisconnected);
 		this.stream.send('disconnect', { id: this.id });
+		this.stream.removeSharedConnectionPool(this);
 	}
 }
 

src/server/api/stream/channel.ts

@@ -7,6 +7,8 @@ import Connection from '.';
 export default abstract class Channel {
 	protected connection: Connection;
 	public id: string;
+	public abstract readonly chName: string;
+	public static readonly shouldShare: boolean;
 
 	protected get user() {
 		return this.connection.user;

src/server/api/stream/channels/drive.ts

@@ -2,6 +2,9 @@ import autobind from 'autobind-decorator';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'drive';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		// Subscribe drive stream

src/server/api/stream/channels/games/reversi-game.ts

@@ -8,6 +8,9 @@ import * as maps from '../../../../../games/reversi/maps';
 import Channel from '../../channel';
 
 export default class extends Channel {
+	public readonly chName = 'gamesReversiGame';
+	public static shouldShare = false;
+
 	private gameId: mongo.ObjectID;
 
 	@autobind

src/server/api/stream/channels/games/reversi.ts

@@ -5,6 +5,9 @@ import { publishMainStream } from '../../../../../stream';
 import Channel from '../../channel';
 
 export default class extends Channel {
+	public readonly chName = 'gamesReversi';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		// Subscribe reversi stream

src/server/api/stream/channels/global-timeline.ts

@@ -5,6 +5,9 @@ import shouldMuteThisNote from '../../../../misc/should-mute-this-note';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'globalTimeline';
+	public static shouldShare = true;
+
 	private mutedUserIds: string[] = [];
 
 	@autobind

src/server/api/stream/channels/hashtag.ts

@@ -5,6 +5,9 @@ import shouldMuteThisNote from '../../../../misc/should-mute-this-note';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'hashtag';
+	public static shouldShare = false;
+
 	@autobind
 	public async init(params: any) {
 		const mute = this.user ? await Mute.find({ muterId: this.user._id }) : null;

src/server/api/stream/channels/home-timeline.ts

@@ -5,6 +5,9 @@ import shouldMuteThisNote from '../../../../misc/should-mute-this-note';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'homeTimeline';
+	public static shouldShare = true;
+
 	private mutedUserIds: string[] = [];
 
 	@autobind

src/server/api/stream/channels/hybrid-timeline.ts

@@ -5,6 +5,9 @@ import shouldMuteThisNote from '../../../../misc/should-mute-this-note';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'hybridTimeline';
+	public static shouldShare = true;
+
 	private mutedUserIds: string[] = [];
 
 	@autobind

src/server/api/stream/channels/local-timeline.ts

@@ -5,6 +5,9 @@ import shouldMuteThisNote from '../../../../misc/should-mute-this-note';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'localTimeline';
+	public static shouldShare = true;
+
 	private mutedUserIds: string[] = [];
 
 	@autobind

src/server/api/stream/channels/main.ts

@@ -3,6 +3,9 @@ import Mute from '../../../../models/mute';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'main';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		const mute = await Mute.find({ muterId: this.user._id });

src/server/api/stream/channels/messaging-index.ts

@@ -2,6 +2,9 @@ import autobind from 'autobind-decorator';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'messagingIndex';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		// Subscribe messaging index stream

src/server/api/stream/channels/messaging.ts

@@ -3,6 +3,9 @@ import read from '../../common/read-messaging-message';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'messaging';
+	public static shouldShare = false;
+
 	private otherpartyId: string;
 
 	@autobind

src/server/api/stream/channels/notes-stats.ts

@@ -5,6 +5,9 @@ import Channel from '../channel';
 const ev = new Xev();
 
 export default class extends Channel {
+	public readonly chName = 'notesStats';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		ev.addListener('notesStats', this.onStats);

src/server/api/stream/channels/server-stats.ts

@@ -5,6 +5,9 @@ import Channel from '../channel';
 const ev = new Xev();
 
 export default class extends Channel {
+	public readonly chName = 'serverStats';
+	public static shouldShare = true;
+
 	@autobind
 	public async init(params: any) {
 		ev.addListener('serverStats', this.onStats);

src/server/api/stream/channels/user-list.ts

@@ -2,6 +2,9 @@ import autobind from 'autobind-decorator';
 import Channel from '../channel';
 
 export default class extends Channel {
+	public readonly chName = 'userList';
+	public static shouldShare = false;
+
 	@autobind
 	public async init(params: any) {
 		const listId = params.listId as string;

src/server/api/stream/index.ts

@@ -148,7 +148,7 @@ export default class Connection {
 	private onChannelConnectRequested(payload: any) {
 		const { channel, id, params } = payload;
 		log(`CH CONNECT: ${id} ${channel} by @${this.user.username}`);
-		this.connectChannel(id, params, (channels as any)[channel]);
+		this.connectChannel(id, params, channel);
 	}
 
 	/**
@@ -177,10 +177,15 @@ export default class Connection {
 	 * チャンネルに接続
 	 */
 	@autobind
-	public connectChannel(id: string, params: any, channelClass: { new(id: string, connection: Connection): Channel }) {
-		const channel = new channelClass(id, this);
-		this.channels.push(channel);
-		channel.init(params);
+	public connectChannel(id: string, params: any, channel: string) {
+		// 共有可能チャンネルに接続しようとしていて、かつそのチャンネルに既に接続していたら無意味なので無視
+		if ((channels as any)[channel].shouldShare && this.channels.some(c => c.chName === channel)) {
+			return;
+		}
+
+		const ch: Channel = new (channels as any)[channel](id, this);
+		this.channels.push(ch);
+		ch.init(params);
 		this.sendMessageToWs('connected', {
 			id: id
 		});

src/server/api/streaming.ts

@@ -6,7 +6,6 @@ import Xev from 'xev';
 import MainStreamConnection from './stream';
 import { ParsedUrlQuery } from 'querystring';
 import authenticate from './authenticate';
-import channels from './stream/channels';
 import { EventEmitter } from 'events';
 import config from '../../config';
 
@@ -66,13 +65,13 @@ module.exports = (server: http.Server) => {
 			};
 
 			main.connectChannel(Math.random().toString(), null,
-				request.resourceURL.pathname === '/' ? channels.homeTimeline :
-				request.resourceURL.pathname === '/local-timeline' ? channels.localTimeline :
-				request.resourceURL.pathname === '/hybrid-timeline' ? channels.hybridTimeline :
-				request.resourceURL.pathname === '/global-timeline' ? channels.globalTimeline : null);
+				request.resourceURL.pathname === '/' ? 'homeTimeline' :
+				request.resourceURL.pathname === '/local-timeline' ? 'localTimeline' :
+				request.resourceURL.pathname === '/hybrid-timeline' ? 'hybridTimeline' :
+				request.resourceURL.pathname === '/global-timeline' ? 'globalTimeline' : null);
 
 			if (request.resourceURL.pathname === '/') {
-				main.connectChannel(Math.random().toString(), null, channels.main);
+				main.connectChannel(Math.random().toString(), null, 'main');
 			}
 		}
 

src/tools/move-drive-files.ts

@@ -1,15 +1,17 @@
 import * as Minio from 'minio';
 import * as uuid from 'uuid';
-const sequential = require('promise-sequential');
-import DriveFile, { DriveFileChunk, getDriveFileBucket } from '../models/drive-file';
+import * as promiseLimit from 'promise-limit';
+import DriveFile, { DriveFileChunk, getDriveFileBucket, IDriveFile } from '../models/drive-file';
 import DriveFileThumbnail, { DriveFileThumbnailChunk } from '../models/drive-file-thumbnail';
 import config from '../config';
 
+const limit = promiseLimit(16);
+
 DriveFile.find({
 	$or: [{
-		withoutChunks: { $exists: false }
+		'metadata.withoutChunks': { $exists: false }
 	}, {
-		withoutChunks: false
+		'metadata.withoutChunks': false
 	}],
 	'metadata.deletedAt': { $exists: false }
 }, {
@@ -19,59 +21,63 @@ DriveFile.find({
 }).then(async files => {
 	console.log(`there is ${files.length} files`);
 
-	await sequential(files.map(file => async () => {
-		file = await DriveFile.findOne({ _id: file._id });
+	await Promise.all(files.map(file => limit(() => job(file))));
 
-		const minio = new Minio.Client(config.drive.config);
-
-		const name = file.filename;
-		const keyDir = `${config.drive.prefix}/${uuid.v4()}`;
-		const key = `${keyDir}/${name}`;
-		const thumbnailKeyDir = `${config.drive.prefix}/${uuid.v4()}`;
-		const thumbnailKey = `${thumbnailKeyDir}/${name}.thumbnail.jpg`;
-
-		const baseUrl = config.drive.baseUrl
-			|| `${ config.drive.config.useSSL ? 'https' : 'http' }://${ config.drive.config.endPoint }${ config.drive.config.port ? `:${config.drive.config.port}` : '' }/${ config.drive.bucket }`;
-
-		const bucket = await getDriveFileBucket();
-		const readable = bucket.openDownloadStream(file._id);
-
-		await minio.putObject(config.drive.bucket, key, readable, file.length, {
-			'Content-Type': file.contentType,
-			'Cache-Control': 'max-age=31536000, immutable'
-		});
-
-		await DriveFile.findOneAndUpdate({ _id: file._id }, {
-			$set: {
-				'metadata.withoutChunks': true,
-				'metadata.storage': 'minio',
-				'metadata.storageProps': {
-					key: key,
-					thumbnailKey: thumbnailKey
-				},
-				'metadata.url': `${ baseUrl }/${ keyDir }/${ encodeURIComponent(name) }`,
-			}
-		});
-
-		// チャンクをすべて削除
-		await DriveFileChunk.remove({
-			files_id: file._id
-		});
-
-		//#region サムネイルもあれば削除
-		const thumbnail = await DriveFileThumbnail.findOne({
-			'metadata.originalId': file._id
-		});
-
-		if (thumbnail) {
-			await DriveFileThumbnailChunk.remove({
-				files_id: thumbnail._id
-			});
-
-			await DriveFileThumbnail.remove({ _id: thumbnail._id });
-		}
-		//#endregion
-
-		console.log('done', file._id);
-	}));
+	console.log('ALL DONE');
 });
+
+async function job(file: IDriveFile): Promise<any> {
+	file = await DriveFile.findOne({ _id: file._id });
+
+	const minio = new Minio.Client(config.drive.config);
+
+	const name = file.filename;
+	const keyDir = `${config.drive.prefix}/${uuid.v4()}`;
+	const key = `${keyDir}/${name}`;
+	const thumbnailKeyDir = `${config.drive.prefix}/${uuid.v4()}`;
+	const thumbnailKey = `${thumbnailKeyDir}/${name}.thumbnail.jpg`;
+
+	const baseUrl = config.drive.baseUrl
+		|| `${ config.drive.config.useSSL ? 'https' : 'http' }://${ config.drive.config.endPoint }${ config.drive.config.port ? `:${config.drive.config.port}` : '' }/${ config.drive.bucket }`;
+
+	const bucket = await getDriveFileBucket();
+	const readable = bucket.openDownloadStream(file._id);
+
+	await minio.putObject(config.drive.bucket, key, readable, file.length, {
+		'Content-Type': file.contentType,
+		'Cache-Control': 'max-age=31536000, immutable'
+	});
+
+	await DriveFile.findOneAndUpdate({ _id: file._id }, {
+		$set: {
+			'metadata.withoutChunks': true,
+			'metadata.storage': 'minio',
+			'metadata.storageProps': {
+				key: key,
+				thumbnailKey: thumbnailKey
+			},
+			'metadata.url': `${ baseUrl }/${ keyDir }/${ encodeURIComponent(name) }`,
+		}
+	});
+
+	// チャンクをすべて削除
+	await DriveFileChunk.remove({
+		files_id: file._id
+	});
+
+	//#region サムネイルもあれば削除
+	const thumbnail = await DriveFileThumbnail.findOne({
+		'metadata.originalId': file._id
+	});
+
+	if (thumbnail) {
+		await DriveFileThumbnailChunk.remove({
+			files_id: thumbnail._id
+		});
+
+		await DriveFileThumbnail.remove({ _id: thumbnail._id });
+	}
+	//#endregion
+
+	console.log('done', file._id);
+}