✌️

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

@@ -1,6 +1,8 @@
 import { EventEmitter } from 'eventemitter3';
+import * as uuid from 'uuid';
 import * as ReconnectingWebsocket from 'reconnecting-websocket';
 import { apiUrl } from '../../../config';
+import MiOS from '../../mios';
 
 /**
  * Misskey stream connection
@@ -8,9 +10,21 @@ import { apiUrl } from '../../../config';
 export default class Connection extends EventEmitter {
 	public state: string;
 	private buffer: any[];
-	private socket: ReconnectingWebsocket;
+	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;
+	private os: MiOS;
 
-	constructor(endpoint, params?) {
+	constructor(os: MiOS, endpoint, params?) {
 		super();
 
 		//#region BIND
@@ -21,6 +35,9 @@ export default class Connection extends EventEmitter {
 		this.close =     this.close.bind(this);
 		//#endregion
 
+		this.id = uuid();
+		this.os = os;
+		this.name = endpoint;
 		this.state = 'initializing';
 		this.buffer = [];
 
@@ -35,6 +52,9 @@ export default class Connection extends EventEmitter {
 		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);
 	}
 
 	/**
@@ -44,11 +64,18 @@ export default class Connection extends EventEmitter {
 		this.state = 'connected';
 		this.emit('_connected_');
 
+		this.connectedAt = new Date();
+
 		// バッファーを処理
 		const _buffer = [].concat(this.buffer); // Shallow copy
 		this.buffer = []; // Clear buffer
-		_buffer.forEach(message => {
-			this.send(message); // Resend each buffered messages
+		_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 });
+			}
 		});
 	}
 
@@ -64,6 +91,11 @@ export default class Connection extends EventEmitter {
 	 * Callback of when received a message from connection
 	 */
 	private onMessage(message) {
+		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);
@@ -75,20 +107,26 @@ export default class Connection extends EventEmitter {
 	/**
 	 * Send a message to connection
 	 */
-	public send(message) {
+	public send(data) {
 		// まだ接続が確立されていなかったらバッファリングして次に接続した時に送信する
 		if (this.state != 'connected') {
-			this.buffer.push(message);
+			this.buffer.push(data);
 			return;
 		}
 
-		this.socket.send(JSON.stringify(message));
+		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);
 	}