DEV Community: Itayp1

NodeJs RabbitMQ - Basic Tutorial (working with queue)

Itayp1 — Tue, 14 Jul 2020 20:18:08 +0000

so why do we use MQ

let say that you have 20000 customers and you want to send to each one email everyday
and the email is very importent

one option is make program that loop through the customers emails and send the mail by smtp or an api

but what will hapend if email service was unaviable for 1 minutes
a lot of customers does not get the email

and lets say that before sending the email we should do some working before
now we need to think how to manage few instances of the program

i sure we can think on creative ways to solve this but why ?!?!
in such cases we can use mq

what is a MQ(message queue)

its holds data until the program pull that data out

one program push an message on a queue
and second program pull message mrom the queue

so how it can help us ?!?!
the queue have a built in macanizem that help us
here a few feature that we get with when we working with a queue

persistence - the massage saved in the queue even if your app will crash
distribution - you can add 20,000 messages in a queue and have another 10 instances that each one pull messages from the queue
fail-safe - if your application crashed the the middle of the process , the message return to the queue and be picked by another instance
and there is more reasons

ok lets start do some coding

in this tutorial we will work with rabbitmq
its open source and very popular
the easiest way to geting start is to register to this site
cloudamqp.com
and create a new instance of mq
alternatively you can install mq in your computer from this link
link

Program one this program responsible for pushing messages into the queue

first you sould get the url of the mq
if you registerd to the site above you can find the connection string in here

and if you installed the mq locally you can just put "localhost" in the connection string

few thing about the code below

amqp.connect - creating connection with the mq server
connection.createConfirmChannel -tell the mq to make sure that each message saved on the mq
channel.assertQueue - checks for the queue, if it doesn't exist then it will create one.
channel.sendToQueue - send message to the queue
- the first argument is the queue name (you decide whats the name of the queue)
- the second argument is the message that we send to the queue you can see that the messages sended in binary to the queue in this case i sended the index i but you can replaced the i with json object or anything else
the object { durable: true } and {persistent: true}

responsible for the message persistent
that in case that the mq server restarted
the messages that was in the queue will stay

in this code we pushed 20 messages to the queue

var amqp = require('amqplib/callback_api');

amqp.connect(`amqp://localhost`, function (error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createConfirmChannel(function (error1, channel) {
        if (error1) {
            throw error1;
        }
        channel.assertQueue(queue, {
            durable: true
        });
        var queue = 'task_queue1';
        for (i = 0; i < 20; i++)
            channel.sendToQueue(queue, Buffer.from(i.toString()),{
                persistent: true
            }, (err, ok) => {
                if (err) {
                    console.log(err)
                } else {
                    console.log(ok)
                }
            });
        console.log(" [x] Sent %s", msg);
    });
});

Program two - The Worker - this program responsible for pulling messages fromthe queue

few thing about the code below

channel.prefetch - the number of messages that the queue can handle every given time
channel.consume -this operation pulling the messages from the queue
- the first argument is the queue name (should be the same name as the queue name in program one)
- channel.ack - after that we read the message from the queue and we finished the processing of this message we send to the mq acknowledge that he could remove this message from the queue
- in the callback we get the message in object msg.content
noAck: false - set it to false otherwise the mq will remove the message from the queue before you send the acknowledge

**the setTimeout simulate heavy processing
its wait 7 seconds before sending the a acknowledge

var amqp = require('amqplib/callback_api');
amqp.connect(`amqp://localhost`, function (error0, connection) {
    if (error0) {
        throw error0
    }
    connection.createChannel(function (error1, channel) {
        if (error1) {
            throw error1;
        }
        var queue = 'task_queue1';
        channel.assertQueue(queue, {
            durable: true
        });
        channel.prefetch(10)
        channel.consume(queue, function (msg) {
            console.log(" [x] Received %s", 
            msg.content.toString());
            setTimeout(function () {
                console.log(" [x] Done");
                channel.ack(msg);
            }, 7000);

        }, {
            noAck: false
        });
    });
});

thats it
i hope that this article will help you to get started with rabbitmq
and understand when its should be in use

javascript- working with stream and large files

Itayp1 — Mon, 13 Jul 2020 19:56:05 +0000

hey
in most cases its ok to read file into the memory
but when the size of the file get greater then the memory usage on the os get greater

assumingthat we working with a server and we recived 50 request simutensly and each request size is 20mb

the the os memory usage jump to 20*50= 1000m
in that way the os run out of memory and the apliication will crash

in such cases we should working with streams
the stream allowed us to get the file in pieces(chunks)
just like Iterate through array

the stream based on events
and we have the following events

//data get the current "pice of the file"
source.on('data', function (chunk) {
});
//end will start after the all file passed through the data event 
source.on('end', function () {
});
//will rise in case of an error 
source.on('error', function (err) {
});

to set the stream we use the common fs

 const read = fs.createReadStream("bigfile.txt")
 const write = fs.createWriteStream("bigfile.txt")

here is example of the whole process

for this example the size of the file bigfile.txt is 50 mb

//in this case we loaded 50 mb into the memory 
const file= fs.readFileSync("bigfile.txt", "utf8")

// in this case we read only one pice from the file in every given time
const source = fs.createReadStream("bigfile.txt", "utf8")
//after that we set the stream variable we can start geting the file data
source.on('data', function (chunk) {
    console.log(chunk)
});
source.on('end', function () {
    console.log("end");
});
source.on('error', function (err) {
    console.log("error" + err);//cant find file or something like that
});

so in this example you can read file even if the request / file size is 5GB
and the memory does not jump at all

and if you want to write into file its pretty much the same

const destination = fs.createWriteStream("bigfile2.txt")
destination.write(chunk)
//and in the end we will close the stream
destination.end()
//and we have the finish and the error event just like the Example above

now lets combine the read and the write


const source = fs.createReadStream("bigfile.txt", "utf8")
const destination = fs.createWriteStream("bigfile2.txt")

source.on('data', function (chunk) {
//write into the file  one piece at a time
    destination.write(chunk)
});
source.on('end', function () {
    //after that we read the all file piece  by piece we close the stram 
    destination.end()

});


destination.on("finish", () => {
//the function destination.end() will rise the finish event 
    console.log("done write into bigfile2.txt")
})

after that we know to working with files
we can implement the stream on other operations
let say we want to read file , compress the data and write the compressed data into a new file

for that we use the liberary zlib and we use the pipeline
pipeline take the readble stream from one side
pass the data through some kind of middleware and then take the output from the middleware and passed it to to destination stream

so in this example we will read a file
compressed him and write him into a new file

const { pipeline } = require('stream');
const { createGzip} = require('zlib');
const gzip = createGzip();

const source = fs.createReadStream("bigfile.txt", "utf8")
const destination = fs.createWriteStream("bigfile3.zip")

pipeline(source, gzip, destination, (err) => {
    if (err) {
        console.error('An error occurred:', err);
    }
});

thats it

javascript combine few functions into pipeline

Itayp1 — Sat, 11 Jul 2020 20:58:35 +0000

combine few functions into pipeline

sometimes we need to pass our data in some pipe line that will produce the updated data

and its can be accomplished in many ways

the simplesd way is that

const obj = { name: "itay", age: 28, isValid: null }

const upperCase = (({ name, ...other }) => ({ ...other, name: name.toUpperCase() }))
const setAge = (({ age, ...other }) => ({ ...other, age: (new Date().getFullYear() - age) }))
const validName = (({ isValid, name, ...other }) => ({ ...other, name, isValid: name.match(/^[A-Za-z]+$/) ? true : false }))
const func1res = upperCase(obj)
const func2res = setAge(func1res)
const func3res = validName(func2res)
console.log(func3res)

yes its working
but its not the best way

we can combine all the funtions into one line like that

console.log(validName(setAge(upperCase(obj))))

but its will become harder to maintain in that way
there is another option
slightly different

reducers

reducer is an method that belong to array object
and its can help us to combine those funtions

i assume that you already know that is an reducer
and if not i will show a little example that sum the number in the array

var numbers = [1, 2, 3, 4];
const sum2Numbers = (accumulator, currentNum) => currentNum + accumulator


console.log(numbers.reduce(sum2Numbers, 0))

this is how we combine the functions with reducer

const compose = (arr, val) => arr.reduce((updatedObj, currentFunc) => currentFunc(updatedObj), val)

its will get the same result
{ age: 1992, name: 'ITAY', isValid: true }

to explain this i will use a difrent example
we have 2 function 2415

const doubleit = (a) => a * a
const addit = (a) => a + a

we want a way to combine thoes funcs with minimum effort
lets begin

first we create function that run on every function and every time get the last calculated value

const combineFumc = (value, currentFunc) => currentFunc(value)
// like the sum2Numbers , in runtime its will be something like
(value, currentFunc) => currentFunc(value)
(2, doubleit ) => doubleit (2)

then we create the compose function

const coompose = (functionList) => (val) => functionList.reduce(combineFumc, val)

the first value will be an array

const additTwiceAndDoubleif = coompose([addit, addit, doubleit, doubleit])

so you can decide how do you want to combine your function

and then you just pass to the function the value to calculate

 console.log(additTwiceAndDoubleif(2))

javascript - how to pass array into chain of functions

Itayp1 — Sat, 11 Jul 2020 19:39:23 +0000

chain functions

lets asume that we have an array of object
and you want to make few operation on the array like
filter , map , sort
so one option will be straightforward
run each time on every operation

const arr = [{ name: itay, age: 10 }, { name: sam, age: 20 }, { name: jett, age: 30 }, { name: bob, age: 40 }]


//filter  the obkect with age under 10
const filtering = ({ age }) => age > 10
filterdArr = arr.filter(filtering)

// abount the sorting - if the function return negetive value then a will be befora b
//and if the funtion  return positive number then b will be before a 
const sorting = (a, b) => a.age - b.age
sortedArr = filterdArr.sort(sorting)

//refactor the object , changed age to myAge
const mapping = ({ name, age }) => ({ name, myAge: age })
mapedArr = sortedArr.map(mapping)

console.log(mapedArr)

the second option will be to chain the funtions
like that

console.log(
    arr.filter(filtering)
        .sort(sorting)
        .map(mapping))

and the result will be the same
[
{ name: 'jett', myAge: 30 },
{ name: 'bob', myAge: 40 },
{ name: 'sam', myAge: 55 }
]

Testitayp123

Itayp1 — Sat, 04 Jul 2020 14:41:40 +0000

Test