How to use the rhea.on function in rhea

var sendable = function (context) {
          clearTimeout(timer);
          this.version = this.connection.properties
            ? this.connection.properties.version
            : "0.1.0";
          // in case this connection dies
          rhea.on("disconnected", this.on_disconnected);
          // in case this connection dies and is then reconnected automatically
          rhea.on("connection_open", this.on_connection_open);
          // receive messages here
          this.connection.on("message", this.on_message);
          resolve(context);
        }.bind(this);
        this.connection.once("sendable", sendable);

var connect_to_enmasse = function (server, address, count, username, password, protocols) {
        var requests = parseInt(count);
        var current = 1;
        var sender;
        var client = require('rhea');

        function next_request() {
            var msg = 'request-' + current;
            sender.send({body: msg});
            append('sent: ' + msg);
        }

        client.on('receiver_open', function (context) {
            next_request();
        });

        function append(txt) {
            var node = document.createTextNode(txt);
            var div = document.createElement("div");
            div.appendChild(node);
            document.body.appendChild(div);
        }

        client.on("message", function (context) {
            append('received: ' + context.message.body);
            if (current++ < requests) {
                next_request();
            } else {
                context.connection.close();

if (args.idle_time_out) {
    connect_options.idle_time_out=args.idle_time_out;
}

function next_request() {
    var msg = 'request-' + current;
    sender.send({body:msg})
    console.log('sent ' + msg);
}

container.on('connection_open', function (context) {
    next_request();
});

container.on('disconnected', function (context) {
    console.log('disconnected');
});

container.on('message', function (context) {
    console.log('received ' + context.message.body);
    if (current++ < requests) {
        timer_task = setTimeout(next_request, args.request_interval);
    } else {
        sender = undefined;
        if (timer_task) clearTimeout(timer_task);
        context.connection.close();
        console.log('connection closed');
    }
});

var connection = container.connect(connect_options);

*
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
var container = require('rhea');
var args = require('../options.js').options({
    'h': { alias: 'host', default: 'localhost', describe: 'dns or ip name of server where you want to connect'},
    'p': { alias: 'port', default: 5672, describe: 'port to listen on'}
}).help('help').argv;

container.sasl_server_mechanisms.enable_anonymous();
var server = container.listen({ port: args.port, host: args.host });
container.on('connection_open', function (context) {
    console.log('Connected!');
});

function next_request(context) {
    if (context.receiver.source.address) {
        sender.send({reply_to:context.receiver.source.address, body:requests[0]})
    }
}

container.on('connection_open', function (context) {
    sender = context.connection.open_sender(args.node);
    context.connection.open_receiver({source:{dynamic:true}});
});
container.on('receiver_open', function (context) {
    next_request(context);
});

container.on('message', function (context) {
    console.log(requests.shift() + " => " + context.message.body);
    if (requests.length) {
        next_request(context);
    } else {
        context.connection.close();
    }
});

container.connect({'port':args.port});

function subscribe(name, sender) {
    listeners[name] = sender;
}

function unsubscribe(name) {
    delete listeners[name];
    if (Object.getOwnPropertyNames(listeners).length === 0) {
        server.close();
    }
}

container.on('sender_open', function (context) {
    subscribe(context.connection.container_id, context.sender);
});
container.on('sender_close', function (context) {
    unsubscribe(context.connection.container_id);
});
container.on('connection_close', function (context) {
    unsubscribe(context.connection.container_id);
});
container.on('disconnected', function (context) {
    unsubscribe(context.connection.container_id);
});

container.on('message', function (context) {
    if (expected === 0 || received < expected) {
        var name = context.connection.container_id;
        console.log('echoed ' + context.message.body + ' to ' + name);
        listeners[name].send(context.message);
        if (++received === expected) {
            context.receiver.detach();

transport: 'tls',
    key: fs.readFileSync(path.resolve(cert_dir, 'tls.key')),
    cert: fs.readFileSync(path.resolve(cert_dir, 'tls.crt'))
});

console.log('Listening on port ' + process.env.LISTENPORT);
container.on('connection_open', function (context) {
    var authenticatedIdentity = { 'sub' : context.connection.sasl_transport.username || 'anonymous' };
    var groups = [ "manage" ];
    var properties = context.connection.local.open.properties || {};
    properties["authenticated-identity"] = authenticatedIdentity;
    properties["groups"] = groups;
    context.connection.local.open.properties = properties;
    context.connection.close();
});
container.on('disconnected', function (context) {
});

var listeners = {};

var server = container.listen({ port: args.port });

function subscribe(name, sender) {
    listeners[name] = sender;
}

function unsubscribe(name) {
    delete listeners[name];
    if (Object.getOwnPropertyNames(listeners).length === 0) {
        server.close();
    }
}

container.on('sender_open', function (context) {
    subscribe(context.connection.container_id, context.sender);
});
container.on('sender_close', function (context) {
    unsubscribe(context.connection.container_id);
});
container.on('connection_close', function (context) {
    unsubscribe(context.connection.container_id);
});
container.on('disconnected', function (context) {
    unsubscribe(context.connection.container_id);
});

container.on('message', function (context) {
    if (expected === 0 || received < expected) {
        var name = context.connection.container_id;
        console.log('echoed ' + context.message.body + ' to ' + name);

* limitations under the License.
 */
var container = require('rhea');

var args = require('./options.js').options({
      'p': { alias: 'port', default: 5672, describe: 'port to connect to'}
    }).help('help').argv;

//container.sasl_server_mechanisms.enable_anonymous();
var server = container.listen({'port':args.port});

container.on('receiver_open', function (context) {
    context.receiver.set_target({address:context.receiver.remote.attach.target.address});
});

container.on('sender_open', function (context) {
    if (context.sender.source.dynamic) {
        var id = container.generate_uuid();
        context.sender.set_source({address:id});
    }
});

function match_source_address(link, address) {
    return link && link.local && link.local.attach && link.local.attach.source
        && link.local.attach.source.address === address;
}

container.on('message', function (context) {
    var request = context.message;
    var reply_to = request.reply_to;
    var response = {to: reply_to};
    console.log("Received: " + request.body);

* Copyright 2015 Red Hat Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
var container = require('rhea');
container.on('connection_open', function (context) {
    context.connection.open_receiver('examples');
    context.connection.open_sender('examples');
});
container.on('message', function (context) {
    console.log(context.message.body);
    context.connection.close();
});
container.on('sendable', function (context) {
    context.sender.send({body:'Hello World!'});
    context.sender.detach();
});
//container.connect({'port':5672});
container.connect({'host':'messaging-enmasse.34.210.100.115.nip.io','port':443, transport:'tls',
                   servername:'messaging-enmasse.34.210.100.115.nip.io',rejectUnauthorized:false});