view release on metacpan or  search on metacpan

examples/chat/js/beekeeper.js  view on Meta::CPAN

        username:  "guest",
        password:  "guest",
        on_connect: function() {...}
    });

    bkpr.send_notification({
        method: "test.foo",
        params: { foo: "bar" }
    });

    bkpr.call_remote_method({
        method:    "test.bar",
        params:     { foo: "baz" },
        on_success: function(result) {...},
        on_error:   function(error) {...}
    });

    bkpr.accept_notifications({
        method:    "test.foo.*",
        on_receive: function(params) {...}
    });

    bkpr.accept_remote_calls({
        method:    "test.bar",
        on_receive: function(params) {...}
    });
*/

function BeekeeperClient () { return {

    mqtt: null,
    host: null,
    client_id: null,
    response_topic: null,
    request_seq: 1,
    subscr_seq: 1,
    pending_req: {},
    subscr_cb: {},
    subscr_re: {},

    connect: function(args) {

        const This = this;

        if (!this.client_id) this._generate_client_id();

        if ('debug' in args) this.debug(args.debug);

        this._debug(`Connecting to MQTT broker at ${args.url}`);

        // It is possible to iterate over a list of servers specifying:
        // url: [{ host: 'localhost', port: 1883 }, ... ]

        // Connect to MQTT broker using websockets
        this.mqtt = mqtt.connect( args.url, {
            username: args.username || 'guest',
            password: args.password || 'guest',
            clientId: this.client_id,
            protocolVersion: 5,
            clean: true,
            keepalive: 60,
            reconnectPeriod: 1000,
            connectTimeout: 30 * 1000
        });

        this.mqtt.on('connect', function (connack) {
            This.host = This.mqtt.options.host;
            This._debug("Connected to MQTT broker at " + This.host);
            This._create_response_topic();
            if (args.on_connect) args.on_connect(connack.properties);
        });

        this.mqtt.on('reconnect', function () {
            // Emitted when a reconnect starts
            This._debug("Reconnecting...");
        });

        this.mqtt.on('close', function () {
            // Emitted after a disconnection
            This._debug("Disconnected");
        });

        this.mqtt.on('disconnect', function (packet) {
            // Emitted after receiving disconnect packet from broker
            This._debug("Disconnected by broker");
        });

        this.mqtt.on('offline', function () {
            // Emitted when the client goes offline
            This._debug("Client offline");
        });

        this.mqtt.on('error', function (error) {
            // Emitted when the client cannot connect
            This._debug(error);
        });

        this.mqtt.on('message', function (topic, message, packet) {

            let jsonrpc;
            try {
                if (message[0] == 0x78) {
                    // Deflated JSON
                    let json = pako.inflate(message, {to:'string'});
                    jsonrpc = JSON.parse(json);
                }
                else {
                    jsonrpc = JSON.parse(message.toString());
                }
            } catch (e) { throw `Received invalid JSON: ${e}` }
            This._debug(`Got  << ${message}`);

            const subscr_id = packet.properties.subscriptionIdentifier;
            const subscr_cb = This.subscr_cb[subscr_id];

            subscr_cb(jsonrpc, packet.properties);
        });
    },

    _generate_client_id: function() {
        // Generate a random client id (128+ bits)