CODEWITHSUNDEEP
javascriptnode.jsasynchronousapache-kafka
ItayB
ItayB

Reputation: 11357

Node.js async functions

The following code should handle http request by produce a request message to some remote server (kafka broker) and wait for consuming a response for it. when a respond message arrive - it should be returned as an http respond (json or something).

router.get('/status', function(req, res, next) {
    // init the producer
    ...

    // 1st async function
    producer.on('ready', function () {
        // some code for generating payloads (data for a message)
        ...

        // 2nd async function
        producer.send(payloads, function (err, data) {
            // some log of success sending message 
            ...

            // 3rd async function
            consumer.on('message', function (message) {
                // got some response message
                res.send("message: " + message);
            });
        });
    });
});

Can I make these sync together even tow it's not mine?

EDIT: I'll try to be more clear. Consider the following code:

function boo() {
    // part 1 - init some consumer
    console.log("1. finish init");

    // part 2 - This is async function. whenever messages will arrive - this function will be fetched.
    consumer.on('message', function (message) {
        console.log("2. message arrive!");
        return message;
    }

    // part 3
    console.log("3. end function");
    return null;
}

Assume that part 2 happen after 1 second. The output will be:

1. finish init
3. end function
2. message arrive!

while my goal is to wait for the async message (part 2) and return it's value. How can I achieve that?

Upvotes: 0

Views: 252

Answers (2)

ItayB
ItayB

Reputation: 11357

When asking this question, I was complete new to node.js. After watching Philip Roberts video I realized how JavaScript is actually work. Then, I've solved my issue with a global messageArray & a messageId counter. Each user request is saved in messageArray (with its relevant handlers objects for later response). Then, the message is sent via kafka into the internal system components. The user won't get respond until the message arrived back from the system). When a message arrived to kafka consumer (from the system components) - we extract the relevant id and respond back to the relevant user). Here is the code:

var messageId = 0;
var messageArray= [];

router.get('/status', function(req, res, next) {
    var o = {id: messageId, req: req, res: res, next: next};
    messageArray.push(o);
    messageId++;

    // send message with kafka producer into the system internal components - THE MESSAGE CONTAINS THE messageId!
});

consumer.on('message', function (message) {
    // Extract the original messageId from the arrived message and look for it in the messageArray
    var messageId = extractMessageId(message);

    var data = dequeueMessageById(messageId);

    // got some response message
    data.res.send("message: " + message);
});

function dequeueMessageById(messageId) {
    for (var i=0 ; i < messageArray.length ; i++) {
        if (messageArray[i].id == messageId) {
            var messageData = messageArray[i];
            messageArray.splice(index, 1); // remove from array
            return messageData;
        }
    } /* for */

    return null;
}

Upvotes: 1

Thalaivar
Thalaivar

Reputation: 23642

You can use async library. 

async.series([
 fn1,
 fn2
 ], function (err, results) {    
 console.log(results);
});

Or you could use https://github.com/AndyShin/sequenty

var sequenty = require('sequenty'); 

function f1(cb) // cb: callback by sequenty
{
  console.log("I'm f1");
  cb(); // please call this after finshed
}

function f2(cb)
{
  console.log("I'm f2");
  cb();
}

sequenty.run([f1, f2]);

Upvotes: 1

Related Questions