整理した

src/client/app/common/scripts/streaming/channel.ts

@@ -0,0 +1,13 @@
+import Stream from './stream';
+import MiOS from '../../mios';
+
+/**
+ * Channel stream connection
+ */
+export default class Connection extends Stream {
+	constructor(os: MiOS, channelId) {
+		super(os, 'channel', {
+			channel: channelId
+		});
+	}
+}

src/client/app/common/scripts/streaming/drive.ts

@@ -0,0 +1,34 @@
+import Stream from './stream';
+import StreamManager from './stream-manager';
+import MiOS from '../../mios';
+
+/**
+ * Drive stream connection
+ */
+export class DriveStream extends Stream {
+	constructor(os: MiOS, me) {
+		super(os, 'drive', {
+			i: me.account.token
+		});
+	}
+}
+
+export class DriveStreamManager extends StreamManager<DriveStream> {
+	private me;
+	private os: MiOS;
+
+	constructor(os: MiOS, me) {
+		super();
+
+		this.me = me;
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new DriveStream(this.os, this.me);
+		}
+
+		return this.connection;
+	}
+}

src/client/app/common/scripts/streaming/home.ts

@@ -0,0 +1,57 @@
+import * as merge from 'object-assign-deep';
+
+import Stream from './stream';
+import StreamManager from './stream-manager';
+import MiOS from '../../mios';
+
+/**
+ * Home stream connection
+ */
+export class HomeStream extends Stream {
+	constructor(os: MiOS, me) {
+		super(os, '', {
+			i: me.account.token
+		});
+
+		// 最終利用日時を更新するため定期的にaliveメッセージを送信
+		setInterval(() => {
+			this.send({ type: 'alive' });
+			me.account.lastUsedAt = new Date();
+		}, 1000 * 60);
+
+		// 自分の情報が更新されたとき
+		this.on('i_updated', i => {
+			if (os.debug) {
+				console.log('I updated:', i);
+			}
+			merge(me, i);
+		});
+
+		// トークンが再生成されたとき
+		// このままではAPIが利用できないので強制的にサインアウトさせる
+		this.on('my_token_regenerated', () => {
+			alert('%i18n:common.my-token-regenerated%');
+			os.signout();
+		});
+	}
+}
+
+export class HomeStreamManager extends StreamManager<HomeStream> {
+	private me;
+	private os: MiOS;
+
+	constructor(os: MiOS, me) {
+		super();
+
+		this.me = me;
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new HomeStream(this.os, this.me);
+		}
+
+		return this.connection;
+	}
+}

src/client/app/common/scripts/streaming/messaging-index.ts

@@ -0,0 +1,34 @@
+import Stream from './stream';
+import StreamManager from './stream-manager';
+import MiOS from '../../mios';
+
+/**
+ * Messaging index stream connection
+ */
+export class MessagingIndexStream extends Stream {
+	constructor(os: MiOS, me) {
+		super(os, 'messaging-index', {
+			i: me.account.token
+		});
+	}
+}
+
+export class MessagingIndexStreamManager extends StreamManager<MessagingIndexStream> {
+	private me;
+	private os: MiOS;
+
+	constructor(os: MiOS, me) {
+		super();
+
+		this.me = me;
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new MessagingIndexStream(this.os, this.me);
+		}
+
+		return this.connection;
+	}
+}

src/client/app/common/scripts/streaming/messaging.ts

@@ -0,0 +1,20 @@
+import Stream from './stream';
+import MiOS from '../../mios';
+
+/**
+ * Messaging stream connection
+ */
+export class MessagingStream extends Stream {
+	constructor(os: MiOS, me, otherparty) {
+		super(os, 'messaging', {
+			i: me.account.token,
+			otherparty
+		});
+
+		(this as any).on('_connected_', () => {
+			this.send({
+				i: me.account.token
+			});
+		});
+	}
+}

src/client/app/common/scripts/streaming/othello-game.ts

@@ -0,0 +1,11 @@
+import Stream from './stream';
+import MiOS from '../../mios';
+
+export class OthelloGameStream extends Stream {
+	constructor(os: MiOS, me, game) {
+		super(os, 'othello-game', {
+			i: me ? me.account.token : null,
+			game: game.id
+		});
+	}
+}

src/client/app/common/scripts/streaming/othello.ts

@@ -0,0 +1,31 @@
+import StreamManager from './stream-manager';
+import Stream from './stream';
+import MiOS from '../../mios';
+
+export class OthelloStream extends Stream {
+	constructor(os: MiOS, me) {
+		super(os, 'othello', {
+			i: me.account.token
+		});
+	}
+}
+
+export class OthelloStreamManager extends StreamManager<OthelloStream> {
+	private me;
+	private os: MiOS;
+
+	constructor(os: MiOS, me) {
+		super();
+
+		this.me = me;
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new OthelloStream(this.os, this.me);
+		}
+
+		return this.connection;
+	}
+}

src/client/app/common/scripts/streaming/requests.ts

@@ -0,0 +1,30 @@
+import Stream from './stream';
+import StreamManager from './stream-manager';
+import MiOS from '../../mios';
+
+/**
+ * Requests stream connection
+ */
+export class RequestsStream extends Stream {
+	constructor(os: MiOS) {
+		super(os, 'requests');
+	}
+}
+
+export class RequestsStreamManager extends StreamManager<RequestsStream> {
+	private os: MiOS;
+
+	constructor(os: MiOS) {
+		super();
+
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new RequestsStream(this.os);
+		}
+
+		return this.connection;
+	}
+}

src/client/app/common/scripts/streaming/server.ts

@@ -0,0 +1,30 @@
+import Stream from './stream';
+import StreamManager from './stream-manager';
+import MiOS from '../../mios';
+
+/**
+ * Server stream connection
+ */
+export class ServerStream extends Stream {
+	constructor(os: MiOS) {
+		super(os, 'server');
+	}
+}
+
+export class ServerStreamManager extends StreamManager<ServerStream> {
+	private os: MiOS;
+
+	constructor(os: MiOS) {
+		super();
+
+		this.os = os;
+	}
+
+	public getConnection() {
+		if (this.connection == null) {
+			this.connection = new ServerStream(this.os);
+		}
+
+		return this.connection;
+	}
+}

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

@@ -0,0 +1,108 @@
+import { EventEmitter } from 'eventemitter3';
+import * as uuid from 'uuid';
+import Connection from './stream';
+
+/**
+ * ストリーム接続を管理するクラス
+ * 複数の場所から同じストリームを利用する際、接続をまとめたりする
+ */
+export default abstract class StreamManager<T extends Connection> extends EventEmitter {
+	private _connection: T = null;
+
+	private disposeTimerId: any;
+
+	/**
+	 * コネクションを必要としているユーザー
+	 */
+	private users = [];
+
+	protected set connection(connection: T) {
+		this._connection = connection;
+
+		if (this._connection == null) {
+			this.emit('disconnected');
+		} else {
+			this.emit('connected', this._connection);
+
+			this._connection.on('_connected_', () => {
+				this.emit('_connected_');
+			});
+
+			this._connection.on('_disconnected_', () => {
+				this.emit('_disconnected_');
+			});
+
+			this._connection.user = 'Managed';
+		}
+	}
+
+	protected get connection() {
+		return this._connection;
+	}
+
+	/**
+	 * コネクションを持っているか否か
+	 */
+	public get hasConnection() {
+		return this._connection != null;
+	}
+
+	public get state(): string {
+		if (!this.hasConnection) return 'no-connection';
+		return this._connection.state;
+	}
+
+	/**
+	 * コネクションを要求します
+	 */
+	public abstract getConnection(): T;
+
+	/**
+	 * 現在接続しているコネクションを取得します
+	 */
+	public borrow() {
+		return this._connection;
+	}
+
+	/**
+	 * コネクションを要求するためのユーザーIDを発行します
+	 */
+	public use() {
+		// タイマー解除
+		if (this.disposeTimerId) {
+			clearTimeout(this.disposeTimerId);
+			this.disposeTimerId = null;
+		}
+
+		// ユーザーID生成
+		const userId = uuid();
+
+		this.users.push(userId);
+
+		this._connection.user = `Managed (${ this.users.length })`;
+
+		return userId;
+	}
+
+	/**
+	 * コネクションを利用し終わってもう必要ないことを通知します
+	 * @param userId use で発行したユーザーID
+	 */
+	public dispose(userId) {
+		this.users = this.users.filter(id => id != userId);
+
+		this._connection.user = `Managed (${ this.users.length })`;
+
+		// 誰もコネクションの利用者がいなくなったら
+		if (this.users.length == 0) {
+			// また直ぐに再利用される可能性があるので、一定時間待ち、
+			// 新たな利用者が現れなければコネクションを切断する
+			this.disposeTimerId = setTimeout(() => {
+				this.disposeTimerId = null;
+
+				this.connection.close();
+				this.connection = null;
+			}, 3000);
+		}
+	}
+}

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

@@ -0,0 +1,137 @@
+import { EventEmitter } from 'eventemitter3';
+import * as uuid from 'uuid';
+import * as ReconnectingWebsocket from 'reconnecting-websocket';
+import { wsUrl } from '../../../config';
+import MiOS from '../../mios';
+
+/**
+ * Misskey stream connection
+ */
+export default class Connection extends EventEmitter {
+	public state: string;
+	private buffer: any[];
+	public socket: ReconnectingWebsocket;
+	public name: string;
+	public connectedAt: Date;
+	public user: string = null;
+	public in: number = 0;
+	public out: number = 0;
+	public inout: Array<{
+		type: 'in' | 'out',
+		at: Date,
+		data: string
+	}> = [];
+	public id: string;
+	public isSuspended = false;
+	private os: MiOS;
+
+	constructor(os: MiOS, endpoint, params?) {
+		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);
+		//#endregion
+
+		this.id = uuid();
+		this.os = os;
+		this.name = endpoint;
+		this.state = 'initializing';
+		this.buffer = [];
+
+		const query = params
+			? Object.keys(params)
+				.map(k => encodeURIComponent(k) + '=' + encodeURIComponent(params[k]))
+				.join('&')
+			: null;
+
+		this.socket = new ReconnectingWebsocket(`${wsUrl}/${endpoint}${query ? '?' + query : ''}`);
+		this.socket.addEventListener('open', this.onOpen);
+		this.socket.addEventListener('close', this.onClose);
+		this.socket.addEventListener('message', this.onMessage);
+
+		// Register this connection for debugging
+		this.os.registerStreamConnection(this);
+	}
+
+	/**
+	 * Callback of when open connection
+	 */
+	private onOpen() {
+		this.state = 'connected';
+		this.emit('_connected_');
+
+		this.connectedAt = new Date();
+
+		// バッファーを処理
+		const _buffer = [].concat(this.buffer); // Shallow copy
+		this.buffer = []; // Clear buffer
+		_buffer.forEach(data => {
+			this.send(data); // Resend each buffered messages
+
+			if (this.os.debug) {
+				this.out++;
+				this.inout.push({ type: 'out', at: new Date(), data });
+			}
+		});
+	}
+
+	/**
+	 * Callback of when close connection
+	 */
+	private onClose() {
+		this.state = 'reconnecting';
+		this.emit('_disconnected_');
+	}
+
+	/**
+	 * Callback of when received a message from connection
+	 */
+	private onMessage(message) {
+		if (this.isSuspended) return;
+
+		if (this.os.debug) {
+			this.in++;
+			this.inout.push({ type: 'in', at: new Date(), data: message.data });
+		}
+
+		try {
+			const msg = JSON.parse(message.data);
+			if (msg.type) this.emit(msg.type, msg.body);
+		} catch (e) {
+			// noop
+		}
+	}
+
+	/**
+	 * Send a message to connection
+	 */
+	public send(data) {
+		if (this.isSuspended) return;
+
+		// まだ接続が確立されていなかったらバッファリングして次に接続した時に送信する
+		if (this.state != 'connected') {
+			this.buffer.push(data);
+			return;
+		}
+
+		if (this.os.debug) {
+			this.out++;
+			this.inout.push({ type: 'out', at: new Date(), data });
+		}
+
+		this.socket.send(JSON.stringify(data));
+	}
+
+	/**
+	 * Close this connection
+	 */
+	public close() {
+		this.os.unregisterStreamConnection(this);
+		this.socket.removeEventListener('open', this.onOpen);
+		this.socket.removeEventListener('message', this.onMessage);
+	}
+}