<commit_before>/* globals DDPCommon, EV */

class RadioStation extends EV {
	constructor() {
		super();

		this.instances = {};

		Meteor.connection._stream.on('message', (raw_msg) => {
			const msg = DDPCommon.parseDDP(raw_msg);
			if (msg && msg.msg === 'changed' && msg.collection && msg.fields && msg.fields.eventName && msg.fields.args) {
				// console.log(raw_msg);
				msg.fields.args.unshift(msg.fields.eventName);
				msg.fields.args.unshift(msg.collection);
				this.emit.apply(this, msg.fields.args);
			}
		});
	}
}

Meteor.RadioStation = new RadioStation;


Meteor.Radio = class Radio extends EV {
	constructor(name, {useCollection} = {useCollection: false}) {
		if (Meteor.RadioStation.instances[name]) {
			console.warn('Radio instance already exists:', name);
			return Meteor.RadioStation.instances[name];
		}

		super();

		Meteor.RadioStation.instances[name] = this;

		this.name = name;
		this.useCollection = useCollection;
		this.subscriptions = {};

		Meteor.RadioStation.on(this.subscriptionName, (...args) => {
			if (this.subscriptions[args[0]]) {
				super.emit.apply(this, args);
			}
		});
	}

	get subscriptionName() {
		return `stream-${this.name}`;
	}

	unsubscribe(eventName) {
		delete this.subscriptions[eventName];
	}

	subscribe(eventName) {
		return Meteor.subscribe(this.subscriptionName, eventName, this.useCollection, {
			onStop: () => {
				this.unsubscribe(eventName);
			}
		});
	}

	once(eventName, callback) {
		if (!this.subscriptions[eventName]) {
			this.subscriptions[eventName] = {
				subscription: this.subscribe(eventName)
			};
		}

		super(eventName, callback);
	}

	on(eventName, callback) {
		if (!this.subscriptions[eventName]) {
			this.subscriptions[eventName] = {
				subscription: this.subscribe(eventName)
			};
		}

		super(eventName, callback);
	}

	emit(...args) {
		Meteor.call(this.subscriptionName, ...args);
	}<commit_msg>Save last message from each event and add method to get<commit_after>/* globals DDPCommon, EV */

class RadioStation extends EV {
	constructor() {
		super();

		this.instances = {};

		Meteor.connection._stream.on('message', (raw_msg) => {
			const msg = DDPCommon.parseDDP(raw_msg);
			if (msg && msg.msg === 'changed' && msg.collection && msg.fields && msg.fields.eventName && msg.fields.args) {
				// console.log(raw_msg);
				msg.fields.args.unshift(msg.fields.eventName);
				msg.fields.args.unshift(msg.collection);
				this.emit.apply(this, msg.fields.args);
			}
		});
	}
}

Meteor.RadioStation = new RadioStation;


Meteor.Radio = class Radio extends EV {
	constructor(name, {useCollection} = {useCollection: false}) {
		if (Meteor.RadioStation.instances[name]) {
			console.warn('Radio instance already exists:', name);
			return Meteor.RadioStation.instances[name];
		}

		super();

		Meteor.RadioStation.instances[name] = this;

		this.name = name;
		this.useCollection = useCollection;
		this.subscriptions = {};

		Meteor.RadioStation.on(this.subscriptionName, (eventName, ...args) => {
			if (this.subscriptions[eventName]) {
				this.subscriptions[eventName].lastMessage = args;
				super.emit.call(this, eventName, ...args);
			}
		});
	}

	get subscriptionName() {
		return `stream-${this.name}`;
	}

	unsubscribe(eventName) {
		delete this.subscriptions[eventName];
	}

	subscribe(eventName) {
		return Meteor.subscribe(this.subscriptionName, eventName, this.useCollection, {
			onStop: () => {
				this.unsubscribe(eventName);
			}
		});
	}

	getLastMessageFromEvent(eventName) {
		const subscription = this.subscriptions[eventName];
		if (subscription && subscription.lastMessage) {
			return subscription.lastMessage;
		}
	}

	once(eventName, callback) {
		if (!this.subscriptions[eventName]) {
			this.subscriptions[eventName] = {
				subscription: this.subscribe(eventName)
			};
		}

		super(eventName, callback);
	}

	on(eventName, callback) {
		if (!this.subscriptions[eventName]) {
			this.subscriptions[eventName] = {
				subscription: this.subscribe(eventName)
			};
		}

		super(eventName, callback);
	}

	emit(...args) {
		Meteor.call(this.subscriptionName, ...args);
	}
{'fraction_non_alphanumeric': 0.15495363591996095, 'fraction_numerical': 0.0002440214738897023, 'mean_word_length': 3.920768307322929, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '2390020', 'n_tokens_mistral': 1554, 'n_tokens_neox': 1361, 'n_words': 221}