<commit_before>// NPM packages
import * as mqtt from 'mqtt';
import MqttEmitter = require('mqtt-emitter');
// Meteor packages
import { Meteor } from 'meteor/meteor';
import { _ } from 'meteor/underscore';
// My packages
import { topicHandler, topicHandlers } from './topics/index';
import {
    DEFAULT_SOURCE_OPTIONS,
    SourceOptions
} from './sourceOptions';
import { Session } from './session';

export class Source {

    private topic = new MqttEmitter();
    private publisher = null;
    private queue = new Array();

    mqtt: mqtt.Client;

    /**
     * Object that stores publish handlers.
     * Key is publish name, and value is its handler
     */
    publishHandlers: { [name: string]: Function } = {};
    /**
     * Object that stores sessions of things
     * Key is thingId, and value is its session object
     */
    sessions: { [thingId: string]: Session } = {};
    /**
     * Object that stores method handlers
     * Key is name of a method, and value is its handler
     */
    methodHandlers: { [method: string]: Function } = {};

    constructor(brokerUrl: string, options?: SourceOptions) {
        // Overide default options with user defined options
        const extendedOptions = _.extend(DEFAULT_SOURCE_OPTIONS, options);

        // Setup MQTT client options
        const mqttOptions: mqtt.ClientOptions = {
            clientId: extendedOptions.serverId
        }

        // Connect to MQTT broker
        this.mqtt = mqtt.connect(brokerUrl, mqttOptions);

        this.initialize();

        this.registerHandlers();
    }

    publish(name, handler, options?) {
        if (_.has(this.publishHandlers, name))
            throw new Error('Duplicated publishes');

        this.publishHandlers[name] = handler;
    }

    methods(methods: { [name: string]: Function }) {
        for (let method in methods) {
            let handler = methods[method];

            if (typeof handler !== 'function')
                throw new Error(`Type of method '${method}' must be function!`);
            if (_.has(this.methodHandlers, method))
                throw new Error(`A method ${method} is already defined`);

            this.methodHandlers[method] = handler;
        }
    }

    send(topic: string, message: number)
    send(topic: string, message: string)
    send(topic: string, message?: any) {
        if (typeof message === 'number')
            message = message.toString();
        if (typeof message === 'object')
            message = JSON.stringify(message);
        if (typeof message === 'undefined')
            message = '';

        this.queue.push({topic, message});

        if (this.publisher === null) {
            this.publisher = setInterval( () => {
                let obj = this.queue.shift();

                this.mqtt.publish(obj.topic, obj.message);

                if (this.queue.length === 0) {
                    clearInterval(this.publisher);

                    this.publisher = null;
                }
            }, 1);
        }
    }

    getSession(thingId: string): Session {
        // Return existing session if it exists
        if (_.has(this.sessions, thingId))
            return this.sessions[thingId];
        // There is no session exists. Create new session
        let newSession = new Session(thingId, this);
        // Register session
        this.sessions[thingId] = newSession;

        return newSession;
    }

    private initialize() {
        this.mqtt.on('connect', Meteor.bindEnvironment(() => {
            console.log('Source connected to MeteMQ Broker!')
            this.mqtt.subscribe('#');
        }));

        this.mqtt.on('message', Meteor.bindEnvironment((topic: string, message: Buffer) => {
            let payload = message.toString();
            this.topic.emit(topic, payload);
        }));
    }

    // Register every topic handlers in topics/index.ts
    private registerHandlers() {
        for (let topic in topicHandlers)<commit_msg>Interval time is changed<commit_after>// NPM packages
import * as mqtt from 'mqtt';
import MqttEmitter = require('mqtt-emitter');
// Meteor packages
import { Meteor } from 'meteor/meteor';
import { _ } from 'meteor/underscore';
// My packages
import { topicHandler, topicHandlers } from './topics/index';
import {
    DEFAULT_SOURCE_OPTIONS,
    SourceOptions
} from './sourceOptions';
import { Session } from './session';

export class Source {

    private topic = new MqttEmitter();
    private publisher = null;
    private queue = new Array();

    mqtt: mqtt.Client;

    /**
     * Object that stores publish handlers.
     * Key is publish name, and value is its handler
     */
    publishHandlers: { [name: string]: Function } = {};
    /**
     * Object that stores sessions of things
     * Key is thingId, and value is its session object
     */
    sessions: { [thingId: string]: Session } = {};
    /**
     * Object that stores method handlers
     * Key is name of a method, and value is its handler
     */
    methodHandlers: { [method: string]: Function } = {};

    constructor(brokerUrl: string, options?: SourceOptions) {
        // Overide default options with user defined options
        const extendedOptions = _.extend(DEFAULT_SOURCE_OPTIONS, options);

        // Setup MQTT client options
        const mqttOptions: mqtt.ClientOptions = {
            clientId: extendedOptions.serverId
        }

        // Connect to MQTT broker
        this.mqtt = mqtt.connect(brokerUrl, mqttOptions);

        this.initialize();

        this.registerHandlers();
    }

    publish(name, handler, options?) {
        if (_.has(this.publishHandlers, name))
            throw new Error('Duplicated publishes');

        this.publishHandlers[name] = handler;
    }

    methods(methods: { [name: string]: Function }) {
        for (let method in methods) {
            let handler = methods[method];

            if (typeof handler !== 'function')
                throw new Error(`Type of method '${method}' must be function!`);
            if (_.has(this.methodHandlers, method))
                throw new Error(`A method ${method} is already defined`);

            this.methodHandlers[method] = handler;
        }
    }

    send(topic: string, message: number)
    send(topic: string, message: string)
    send(topic: string, message?: any) {
        if (typeof message === 'number')
            message = message.toString();
        if (typeof message === 'object')
            message = JSON.stringify(message);
        if (typeof message === 'undefined')
            message = '';

        this.queue.push({topic, message});

        if (this.publisher === null) {
            this.publisher = setInterval( () => {
                let obj = this.queue.shift();

                this.mqtt.publish(obj.topic, obj.message);

                if (this.queue.length === 0) {
                    clearInterval(this.publisher);

                    this.publisher = null;
                }
            }, 0);
        }
    }

    getSession(thingId: string): Session {
        // Return existing session if it exists
        if (_.has(this.sessions, thingId))
            return this.sessions[thingId];
        // There is no session exists. Create new session
        let newSession = new Session(thingId, this);
        // Register session
        this.sessions[thingId] = newSession;

        return newSession;
    }

    private initialize() {
        this.mqtt.on('connect', Meteor.bindEnvironment(() => {
            console.log('Source connected to MeteMQ Broker!')
            this.mqtt.subscribe('#');
        }));

        this.mqtt.on('message', Meteor.bindEnvironment((topic: string, message: Buffer) => {
            let payload = message.toString();
            this.topic.emit(topic, payload);
        }));
    }

    // Register every topic handlers in topics/index.ts
    private registerHandlers() {
        for (let topic in topicHandlers)
{'fraction_non_alphanumeric': 0.11896660365469439, 'fraction_numerical': 0.0005040957781978576, 'mean_word_length': 2.159235668789809, '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': '6308975', 'n_tokens_mistral': 2174, 'n_tokens_neox': 2088, 'n_words': 614}