#925, #926, and refactoring

src/web/app/common/scripts/stream-manager.ts

@@ -1,33 +0,0 @@
-import * as uuid from 'uuid';
-import Connection from './stream';
-
-export default abstract class StreamManager<T extends Connection> {
-	protected connection: T = null;
-
-	/**
-	 * コネクションを必要としているユーザー
-	 */
-	private users = [];
-
-	public abstract getConnection(): T;
-
-	public use() {
-		// ユーザーID生成
-		const userId = uuid();
-
-		this.users.push(userId);
-
-		return userId;
-	}
-
-	public dispose(userId) {
-		this.users = this.users.filter(id => id != userId);
-
-		// 誰もコネクションの利用者がいなくなったら
-		if (this.users.length == 0) {
-			// コネクションを切断する
-			this.connection.close();
-			this.connection = null;
-		}
-	}
-}

src/web/app/common/scripts/streaming/channel-stream.ts

@@ -3,12 +3,10 @@ import Stream from './stream';
 /**
  * Channel stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor(channelId) {
 		super('channel', {
 			channel: channelId
 		});
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/drive-stream.ts

@@ -3,12 +3,10 @@ import Stream from './stream';
 /**
  * Drive stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor(me) {
 		super('drive', {
 			i: me.token
 		});
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/home-stream-manager.ts

@@ -0,0 +1,20 @@
+import StreamManager from './stream-manager';
+import Connection from './home-stream';
+
+export default class HomeStreamManager extends StreamManager<Connection> {
+	private me;
+
+	constructor(me) {
+		super();
+
+		this.me = me;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new Connection(this.me);
+		}
+
+		return this.connection;
+	}
+}

src/web/app/common/scripts/streaming/home-stream.ts

@@ -1,10 +1,10 @@
 import Stream from './stream';
-import signout from './signout';
+import signout from '../signout';
 
 /**
  * Home stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor(me) {
 		super('', {
 			i: me.token
@@ -15,13 +15,14 @@ class Connection extends Stream {
 			this.send({ type: 'alive' });
 		}, 1000 * 60);
 
+		// 自分の情報が更新されたとき
 		(this as any).on('i_updated', me.update);
 
+		// トークンが再生成されたとき
+		// このままではAPIが利用できないので強制的にサインアウトさせる
 		(this as any).on('my_token_regenerated', () => {
 			alert('%i18n:common.my-token-regenerated%');
 			signout();
 		});
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/messaging-index-stream.ts

@@ -3,12 +3,10 @@ import Stream from './stream';
 /**
  * Messaging index stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor(me) {
 		super('messaging-index', {
 			i: me.token
 		});
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/messaging-stream.ts

@@ -3,7 +3,7 @@ import Stream from './stream';
 /**
  * Messaging stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor(me, otherparty) {
 		super('messaging', {
 			i: me.token,
@@ -17,5 +17,3 @@ class Connection extends Stream {
 		});
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/requests-stream.ts

@@ -3,10 +3,8 @@ import Stream from './stream';
 /**
  * Requests stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor() {
 		super('requests');
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/server-stream.ts

@@ -3,10 +3,8 @@ import Stream from './stream';
 /**
  * Server stream connection
  */
-class Connection extends Stream {
+export default class Connection extends Stream {
 	constructor() {
 		super('server');
 	}
 }
-
-export default Connection;

src/web/app/common/scripts/streaming/stream-manager.ts

@@ -0,0 +1,73 @@
+import { EventEmitter } from 'eventemitter3';
+import * as uuid from 'uuid';
+import Connection from './stream';
+
+/**
+ * ストリーム接続を管理するクラス
+ * 複数の場所から同じストリームを利用する際、接続をまとめたりする
+ */
+export default abstract class StreamManager<T extends Connection> extends EventEmitter {
+	protected _connection: T = null;
+
+	/**
+	 * コネクションを必要としているユーザー
+	 */
+	private users = [];
+
+	protected set connection(connection: T) {
+		this._connection = connection;
+
+		if (this._connection == null) {
+			this.emit('disconnected');
+		} else {
+			this.emit('connected', this._connection);
+		}
+	}
+
+	protected get connection() {
+		return this._connection;
+	}
+
+	/**
+	 * コネクションを持っているか否か
+	 */
+	public get hasConnection() {
+		return this._connection != null;
+	}
+
+	/**
+	 * コネクションを要求します
+	 */
+	public abstract getConnection(): T;
+
+	public borrow() {
+		return this._connection;
+	}
+
+	/**
+	 * コネクションを要求するためのユーザーIDを発行します
+	 */
+	public use() {
+		// ユーザーID生成
+		const userId = uuid();
+
+		this.users.push(userId);
+
+		return userId;
+	}
+
+	/**
+	 * コネクションを利用し終わってもう必要ないことを通知します
+	 * @param userId use で発行したユーザーID
+	 */
+	public dispose(userId) {
+		this.users = this.users.filter(id => id != userId);
+
+		// 誰もコネクションの利用者がいなくなったら
+		if (this.users.length == 0) {
+			// コネクションを切断する
+			this.connection.close();
+			this.connection = null;
+		}
+	}
+}

src/web/app/common/scripts/streaming/stream.ts

@@ -1,25 +1,25 @@
+import { EventEmitter } from 'eventemitter3';
 import * as ReconnectingWebsocket from 'reconnecting-websocket';
-import * as riot from 'riot';
-import CONFIG from './config';
+import CONFIG from '../config';
 
 /**
  * Misskey stream connection
  */
-class Connection {
+export default class Connection extends EventEmitter {
 	private state: string;
 	private buffer: any[];
 	private socket: ReconnectingWebsocket;
 
 	constructor(endpoint, params?) {
-		// BIND -----------------------------------
+		super();
+
+		//#region BIND
 		this.onOpen =    this.onOpen.bind(this);
 		this.onClose =   this.onClose.bind(this);
 		this.onMessage = this.onMessage.bind(this);
 		this.send =      this.send.bind(this);
 		this.close =     this.close.bind(this);
-		// ----------------------------------------
-
-		riot.observable(this);
+		//#endregion
 
 		this.state = 'initializing';
 		this.buffer = [];
@@ -42,7 +42,7 @@ class Connection {
 	 */
 	private onOpen() {
 		this.state = 'connected';
-		(this as any).trigger('_connected_');
+		this.emit('_connected_');
 
 		// バッファーを処理
 		const _buffer = [].concat(this.buffer); // Shallow copy
@@ -57,7 +57,7 @@ class Connection {
 	 */
 	private onClose() {
 		this.state = 'reconnecting';
-		(this as any).trigger('_closed_');
+		this.emit('_closed_');
 	}
 
 	/**
@@ -66,7 +66,7 @@ class Connection {
 	private onMessage(message) {
 		try {
 			const msg = JSON.parse(message.data);
-			if (msg.type) (this as any).trigger(msg.type, msg.body);
+			if (msg.type) this.emit(msg.type, msg.body);
 		} catch (e) {
 			// noop
 		}
@@ -93,5 +93,3 @@ class Connection {
 		this.socket.removeEventListener('message', this.onMessage);
 	}
 }
-
-export default Connection;