rhea/rhea.js
function amqpRequesterNode(config) {

        RED.nodes.createNode(this, config);

        //var container = rhea.create_container();

        // get endpoint configuration
        this.endpoint = RED.nodes.getNode(config.endpoint);
        // get all other configuration
        this.address = config.address;

        var node = this;
        // node not yet connected
        this.status({ fill: 'red', shape: 'dot', text: 'disconnected' });

        if (this.endpoint) {

            node.endpoint.connect(function(connection) {
                setup(connection);
            });

            /**
             * Node setup for creating sender link for sending the request
             * and the dynamic receiver for receiving reply
             *
             * @param connection    Connection instance
             */
            function setup(connection) {

                node.connection = connection;

                // node connected
                node.status({ fill: 'green', shape: 'dot', text: 'connected' });

                // build sender options based on node configuration
                var sender_options = {
                    target: {
                        address: node.address
                    }
                };
                node.sender = node.connection.open_sender(sender_options);

                node.sender.on('accepted', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                node.sender.on('released', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                node.sender.on('rejected', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                // build receiver options
                var receiver_options = {
                    source: {
                        dynamic: true
                    }
                };
                node.receiver = node.connection.open_receiver(receiver_options);

                node.receiver.on('message', function(context) {
                    var msg = {
                        payload: context.message,
                        delivery: context.delivery
                    };
                    node.send([ ,msg]);
                });

                node.connection.on('disconnected', function(context) {
                    // node disconnected
                    node.status({fill: 'red', shape: 'dot', text: 'disconnected' });
                });

                node.on('input', function(msg) {
                    // enough credits to send
                    if (node.sender.sendable()) {

                        if (node.receiver.source.address) {
                            node.sender.send({ reply_to: node.receiver.source.address, body: msg.payload.body });
                        }

                    }
                });

                node.on('close', function() {
                    if (node.sender != null)
                        node.sender.detach();
                    if (node.receiver != null)
                        node.receiver.detach();
                    node.connection.close();
                })
            }

        }
    }
function amqpRequesterNode(config) {

        RED.nodes.createNode(this, config);

        //var container = rhea.create_container();

        // get endpoint configuration
        this.endpoint = RED.nodes.getNode(config.endpoint);
        // get all other configuration
        this.address = config.address;

        var node = this;
        // node not yet connected
        this.status({ fill: 'red', shape: 'dot', text: 'disconnected' });

        if (this.endpoint) {

            node.endpoint.connect(function(connection) {
                setup(connection);
            });

            /**
             * Node setup for creating sender link for sending the request
             * and the dynamic receiver for receiving reply
             *
             * @param connection    Connection instance
             */
            function setup(connection) {

                node.connection = connection;

                // node connected
                node.status({ fill: 'green', shape: 'dot', text: 'connected' });

                // build sender options based on node configuration
                var sender_options = {
                    target: {
                        address: node.address
                    }
                };
                node.sender = node.connection.open_sender(sender_options);

                node.sender.on('accepted', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                node.sender.on('released', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                node.sender.on('rejected', function(context) {
                    var msg = outDelivery(node, context.delivery);
                    node.send([msg, ]);
                });

                // build receiver options
                var receiver_options = {
                    source: {
                        dynamic: true
                    }
                };
                node.receiver = node.connection.open_receiver(receiver_options);

                node.receiver.on('message', function(context) {
                    var msg = {
                        payload: context.message,
                        delivery: context.delivery
                    };
                    node.send([ ,msg]);
                });

                node.connection.on('disconnected', function(context) {
                    // node disconnected
                    node.status({fill: 'red', shape: 'dot', text: 'disconnected' });
                });

                node.on('input', function(msg) {
                    // enough credits to send
                    if (node.sender.sendable()) {

                        if (node.receiver.source.address) {
                            node.sender.send({ reply_to: node.receiver.source.address, body: msg.payload.body });
                        }

                    }
                });

                node.on('close', function() {
                    if (node.sender != null)
                        node.sender.detach();
                    if (node.receiver != null)
                        node.receiver.detach();
                    node.connection.close();
                })
            }

        }
    }
function amqpRequesterNode ( config ) { RED . nodes . createNode ( this , config ) ; //var container = rhea.create_container(); // get endpoint configuration this . endpoint = RED . nodes . getNode ( config . endpoint ) ; // get all other configuration this . address = config . address ; var node = this ; // node not yet connected this . status ( { fill : 'red' , shape : 'dot' , text : 'disconnected' } ) ; if ( this . endpoint ) { node . endpoint . connect ( function ( connection ) { setup ( connection ) ; } ) ; /**
             * Node setup for creating sender link for sending the request
             * and the dynamic receiver for receiving reply
             *
             * @param connection    Connection instance
             */ function setup ( connection ) { node . connection = connection ; // node connected node . status ( { fill : 'green' , shape : 'dot' , text : 'connected' } ) ; // build sender options based on node configuration var sender_options = { target : { address : node . address } } ; node . sender = node . connection . open_sender ( sender_options ) ; node . sender . on ( 'accepted' , function ( context ) { var msg = outDelivery ( node , context . delivery ) ; node . send ( [ msg , ] ) ; } ) ; node . sender . on ( 'released' , function ( context ) { var msg = outDelivery ( node , context . delivery ) ; node . send ( [ msg , ] ) ; } ) ; node . sender . on ( 'rejected' , function ( context ) { var msg = outDelivery ( node , context . delivery ) ; node . send ( [ msg , ] ) ; } ) ; // build receiver options var receiver_options = { source : { dynamic : true } } ; node . receiver = node . connection . open_receiver ( receiver_options ) ; node . receiver . on ( 'message' , function ( context ) { var msg = { payload : context . message , delivery : context . delivery } ; node . send ( [ , msg ] ) ; } ) ; node . connection . on ( 'disconnected' , function ( context ) { // node disconnected node . status ( { fill : 'red' , shape : 'dot' , text : 'disconnected' } ) ; } ) ; node . on ( 'input' , function ( msg ) { // enough credits to send if ( node . sender . sendable ( ) ) { if ( node . receiver . source . address ) { node . sender . send ( { reply_to : node . receiver . source . address , body : msg . payload . body } ) ; } } } ) ; node . on ( 'close' , function ( ) { if ( node . sender != null ) node . sender . detach ( ) ; if ( node . receiver != null ) node . receiver . detach ( ) ; node . connection . close ( ) ; } ) } } }
Node for AMQP requester
Node for AMQP requester