<commit_before>/**
 * Connect an external client / transport through the MHub protocol to the
 * internal Hub.
 */

import * as events from "events";

import log from "./log";

import Hub, { Permissions } from "./hub";
import * as protocol from "./protocol";
import * as pubsub from "./pubsub";
import Message from "./message";
import Dict from "./dict";

type ResponseHandler = (response: protocol.Response) => void;

class SubscriptionNode implements pubsub.Destination {
	public name: string;
	private _conn: HubClient;
	private _id: string;
	private _nodes: pubsub.Source[] = [];
	private _onResponse: ResponseHandler;

	constructor(conn: HubClient, id: string, onResponse: ResponseHandler) {
		this._conn = conn;
		this.name = this._conn.name + "_" + id;
		this._id = id;
		this._onResponse = onResponse;
	}

	public send(message: Message): void {
		log.debug("-> %s", this.name, message.topic);
		const response: protocol.MessageResponse = {
			type: "message",
			topic: message.topic,
			data: message.data,
			headers: message.headers,
			subscription: this._id,
		};
		this._onResponse(response);
	}

	public bind(node: pubsub.Source, pattern?: string): void {
		if (this._nodes.indexOf(node) < 0) {
			this._nodes.push(node);
		}
		node.bind(this, pattern);
	}

	public destroy(): void {
		this._nodes.forEach((node: pubsub.Source): void => {
			node.unbind(this);
		});
		this._nodes = [];
	}
}

export class HubClient extends events.EventEmitter {
	public name: string; // TODO move this to higher layer?
	private _hub: Hub;
	private _subscriptions: Dict<SubscriptionNode> = new Dict<SubscriptionNode>();
	private _username: string;
	private _permissions: Permissions;

	constructor(hub: Hub, name: string) {
		super();
		this._hub = hub;
		this.name = name;
		this._permissions = this._hub.getUserPermissions(""); // permissions for anonymous user
	}

	public close(): void {
		this._subscriptions.forEach((subscription) => subscription.destroy());
		this._subscriptions.clear();
	}

	public setUsername(username: string): void {
		this._username = username;
		this._permissions = this._hub.getUserPermissions(this._username);
	}

	public processCommand(msg: protocol.Command): void {
		let errorMessage: string;
		let response: protocol.Response;
		let haveSeq = false;
		let seq: number = undefined;
		try {
			haveSeq = typeof msg === "object" && typeof msg.seq === "number";
			seq = haveSeq ? msg.seq : undefined;

			if (msg.type === "publish" || msg.type === "subscribe") {
				const node = this._hub.find(msg.node);
				if (!this._permissions[msg.type]) {
					errorMessage = "permission denied";
				} else if (!node) {
					errorMessage = `unknown node '${msg.node}'`;
				} else if (msg.type === "publish") {
					const pubCmd = <protocol.PublishCommand>msg;
					if (!pubsub.isDestination(node)) {
						errorMessage = `node '${msg.node}' is not a Destination`;
					} else {
						node.send(new Message(pubCmd.topic, pubCmd.data, pubCmd.headers));
						if (haveSeq) {
							response = {
								type: "puback",
								seq: seq,
							};
						}
					}
				} else { // msg.type === "subscribe"
					const subCmd = <protocol.SubscribeCommand>msg;
					if (!pubsub.isSource(node)) {
						errorMessage = `node '${msg.node}' is not a Source`;
					} else {
						const id = subCmd.id || "default";
						let sub = this._subscriptions.get(id);
						if (!sub) {
							sub = new SubscriptionNode(this, id, this._onResponseHandler);
							this._subscriptions.set(id, sub);
						}
						sub.bind(node, subCmd.pattern);
						if (haveSeq) {
							response = {
								type: "suback",
								seq: seq,
							};
						}
					}
				}
			} else if (msg.type === "ping") {
				response = <protocol.PingAckResponse>{
					type: "pingack",
					seq: seq,
				};
			} else if (msg.type === "login") {
				if (this._username !== undefined) {
					// Wouldn't really be a problem for now, but may be in
					// the future if e.g. different users have different quota
					// etc.
					errorMessage = "already logged in";
				} else {
					const authenticated = this._hub.authenticate(msg.username, msg.password);
					if (!authenticated) {
						errorMessage = "authentication failed";
					} else {
						this.setUsername(msg.username);
						if (haveSeq) {
							response = <protocol.LoginAckResponse>{
								type: "loginack",
								seq: seq,
							};
						}
					}
				}
			} else {
				errorMessage = `unknown command '${msg!.type}'`;
			}
		} catch (e) {
			errorMessage = "decode error: " + String(e);
		}
		if (errorMessage) {
			log.error(`[ ${this.name} ] error: ${errorMessage}`);
			response = {
				type: "error",
				message: errorMessage,
				seq: seq,
			};
		}
		if (response) {
			this._onResponseHandler(response);
		}
	}

	private _onResponseHandler = (response: protocol.Response): void => {
		this.emit("response", response);
	};
}

export default HubClient;<commit_msg>hubclient: Refactor processCommand() into individual methods per command.<commit_after>/**
 * Connect an external client / transport through the MHub protocol to the
 * internal Hub.
 */

import * as events from "events";

import log from "./log";

import Hub, { Permissions } from "./hub";
import * as protocol from "./protocol";
import * as pubsub from "./pubsub";
import Message from "./message";
import Dict from "./dict";

type ResponseHandler = (response: protocol.Response) => void;

class SubscriptionNode implements pubsub.Destination {
	public name: string;
	private _conn: HubClient;
	private _id: string;
	private _nodes: pubsub.Source[] = [];
	private _onResponse: ResponseHandler;

	constructor(conn: HubClient, id: string, onResponse: ResponseHandler) {
		this._conn = conn;
		this.name = this._conn.name + "_" + id;
		this._id = id;
		this._onResponse = onResponse;
	}

	public send(message: Message): void {
		log.debug("-> %s", this.name, message.topic);
		const response: protocol.MessageResponse = {
			type: "message",
			topic: message.topic,
			data: message.data,
			headers: message.headers,
			subscription: this._id,
		};
		this._onResponse(response);
	}

	public bind(node: pubsub.Source, pattern?: string): void {
		if (this._nodes.indexOf(node) < 0) {
			this._nodes.push(node);
		}
		node.bind(this, pattern);
	}

	public destroy(): void {
		this._nodes.forEach((node: pubsub.Source): void => {
			node.unbind(this);
		});
		this._nodes = [];
	}
}

export class HubClient extends events.EventEmitter {
	public name: string; // TODO move this to higher layer?
	private _hub: Hub;
	private _subscriptions: Dict<SubscriptionNode> = new Dict<SubscriptionNode>();
	private _username: string;
	private _permissions: Permissions;

	constructor(hub: Hub, name: string) {
		super();
		this._hub = hub;
		this.name = name;
		this._permissions = this._hub.getUserPermissions(""); // permissions for anonymous user
	}

	public close(): void {
		this._subscriptions.forEach((subscription) => subscription.destroy());
		this._subscriptions.clear();
	}

	public setUsername(username: string): void {
		this._username = username;
		this._permissions = this._hub.getUserPermissions(this._username);
	}

	public processCommand(msg: protocol.Command): void {
		let response: protocol.Response | undefined;
		try {
			switch (msg.type) {
				case "publish":
					response = this._handlePublish(msg);
					break;
				case "subscribe":
					response = this._handleSubscribe(msg);
					break;
				case "ping":
					response = this._handlePing(msg);
					break;
				case "login":
					response = this._handleLogin(msg);
					break;
				default:
					throw new Error(`unknown command '${msg!.type}'`);
			}
		} catch (e) {
			const errorMessage = String(e);
			if (errorMessage) {
				log.error(`[ ${this.name} ] error: ${errorMessage}`);
				response = {
					type: "error",
					message: errorMessage,
					seq: msg.seq,
				};
			}
		}
		if (response) {
			this._onResponseHandler(response);
		}
	}

	private _hasSequenceNumber(msg: protocol.Command): boolean {
		return typeof msg === "object" && typeof msg.seq === "number";
	}

	private _handlePublish(msg: protocol.PublishCommand): protocol.PubAckResponse | undefined {
		const node = this._hub.find(msg.node);
		if (!node) {
			throw new Error(`unknown node '${msg.node}'`);
		}

		if (!this._permissions[msg.type]) {
			throw new Error("permission denied");
		}

		if (!pubsub.isDestination(node)) {
			throw new Error(`node '${msg.node}' is not a Destination`);
		}

		node.send(new Message(msg.topic, msg.data, msg.headers));

		if (this._hasSequenceNumber(msg)) {
			return {
				type: "puback",
				seq: msg.seq,
			};
		}
	}

	private _handleSubscribe(msg: protocol.SubscribeCommand): protocol.SubAckResponse | undefined {
		const node = this._hub.find(msg.node);
		if (!node) {
			throw new Error(`unknown node '${msg.node}'`);
		}

		if (!this._permissions[msg.type]) {
			throw new Error("permission denied");
		}

		if (!pubsub.isSource(node)) {
			throw new Error(`node '${msg.node}' is not a Source`);
		}

		const id = msg.id || "default";
		let sub = this._subscriptions.get(id);
		if (!sub) {
			sub = new SubscriptionNode(this, id, this._onResponseHandler);
			this._subscriptions.set(id, sub);
		}
		sub.bind(node, msg.pattern);

		if (this._hasSequenceNumber(msg)) {
			return {
				type: "suback",
				seq: msg.seq,
			};
		}
	}

	private _handlePing(msg: protocol.PingCommand): protocol.PingAckResponse {
		return {
			type: "pingack",
			seq: msg.seq,
		};
	}

	private _handleLogin(msg: protocol.LoginCommand): protocol.LoginAckResponse | undefined {
		if (this._username !== undefined) {
			// Wouldn't really be a problem for now, but may be in
			// the future if e.g. different users have different quota
			// etc.
			throw new Error("already logged in");
		}

		const authenticated = this._hub.authenticate(msg.username, msg.password);
		if (!authenticated) {
			throw new Error("authentication failed");
		}
		this.setUsername(msg.username);

		if (this._hasSequenceNumber(msg)) {
			return {
				type: "loginack",
				seq: msg.seq,
			};
		}
	}

	private _onResponseHandler = (response: protocol.Response): void => {
		this.emit("response", response);
	};
}

export default HubClient;
{'fraction_non_alphanumeric': 0.14498724740043162, 'fraction_numerical': 0.00019619383951343929, 'mean_word_length': 3.327249575551783, 'pattern_counts': {'":': 4, '<': 13, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2708629', 'n_tokens_mistral': 3817, 'n_tokens_neox': 3199, 'n_words': 829}