How to process data asynchronously in Elixir?

sahilpaudel

Hi Everyone,

I have controller that takes request with a CSV file & email_id in the parameter.
What is does each read through each line of the CSV file and process the data and insert into the DB and returns the processed data for each line from the DB.

I add this processed data into a list and dump it as iodata into CSV binary and send that CSV file as an attachment using :bamboo library.

The problem I am facing is as the size of the file grows the request is taking time to process so I want to do it asynchronously.

Like I get /process data request and then send response as it is being processed to the client and do the job in the backend and send the mail on completion of the job.

Below is the snippet of my code.

def call(csv_path, email) do
    csv_path
      |> read_csv
      |> dump_to_csv
      |> add_attachment
      |> Email.create(email)
      |> Email.send_email
      |> form_response
  end

  def read_csv(csv_path) do
      csv_path
        |> File.stream!
        |> MyParser.parse_stream
        |> Stream.map(fn [arg1, arg2] -> %{prefix: arg1, url: arg2} end)
        |> Enum.flat_map(fn(params) -> url_convertor(params) end)
  end

  def url_convertor(%{url: url, prefix: prefix}) do
    with {:ok, %Route{} = route} <- MyApp.process_data(%{"prefix" => prefix, "url" => url}) do
      route
        |> MyApp.processed_data
        |> add_to_list(route, prefix)
    else
      {:error, %{changes: changes, errors: _}} -> [[prefix, changes[:url], "INVALID_URL"]] 
    end
  end

  def add_to_list(new_url, %{url: original_url}, prefix) do
    [[prefix, original_url, new_url] | []]
  end

  def dump_to_csv(list) do
    [["PREFIX", "ORIGINAL_URL", "NEW_URL"]] ++ list
      |> MyParser.dump_to_iodata()
      |> IO.iodata_to_binary
  end

  def add_attachment(binary_data) do
    %Bamboo.Attachment{content_type: "text/csv", filename: "data.csv", data: binary_data}
  end

Library Used:

:nimble_csv, bamboo_smtp

Can you help me how to make this process async.

14 comments

/phoenix #learning-elixir #asynchronous

0 2145 14

2020-04-16 01:14:12 UTC

Showing Posts 1 to 10

Show Best Posts
Show All (oldest first)
Show All (newest first)

cpgo

You could try using Task.async

Post #1

sahilpaudel

Hi @cpgo Can you please help me out here as this is new to me ?

Post #2

sahilpaudel

I updated my code to

def call(csv_path, email) do
    task = Task.async(fn -> do_async_task(csv_path, email) end)
    res = send_status() 
    res + Task.await(task)
  end

  def send_status() do
    %{response: %{message: "Request Initiated"}}
  end

  def do_async_task(csv_path, email) do
    csv_path
      |> read_csv
      |> dump_to_csv
      |> add_attachment
      |> Email.create(email)
      |> Email.send_email
  end

But still no luck can you guys help me here?

Post #3

NobbZ

res will be a map, you can’t have a map on the left hand side of +/2.

What are you trying to achieve?

Post #4

sahilpaudel

sahilpaudel:

Hi Everyone,

I have controller that takes request with a CSV file & email_id in the parameter.
What is does each read through each line of the CSV file and process the data and insert into the DB and returns the processed data for each line from the DB.

I add this processed data into a list and dump it as iodata into CSV binary and send that CSV file as an attachment using :bamboo library.

The problem I am facing is as the size of the file grows the request is taking time to process so I want to do it asynchronously.

Like I get /process data request and then send response as it is being processed to the client and do the job in the backend and send the mail on completion of the job.

Below is the snippet of my code.
def call(csv_path, email) do
    csv_path
      |> read_csv
      |> dump_to_csv
      |> add_attachment
      |> Email.create(email)
      |> Email.send_email
      |> form_response
  end

  def read_csv(csv_path) do
      csv_path
        |> File.stream!
        |> MyParser.parse_stream
        |> Stream.map(fn [arg1, arg2] -> %{prefix: arg1, url: arg2} end)
        |> Enum.flat_map(fn(params) -> url_convertor(params) end)
  end

  def url_convertor(%{url: url, prefix: prefix}) do
    with {:ok, %Route{} = route} <- MyApp.process_data(%{"prefix" => prefix, "url" => url}) do
      route
        |> MyApp.processed_data
        |> add_to_list(route, prefix)
    else
      {:error, %{changes: changes, errors: _}} -> [[prefix, changes[:url], "INVALID_URL"]] 
    end
  end

  def add_to_list(new_url, %{url: original_url}, prefix) do
    [[prefix, original_url, new_url] | []]
  end

  def dump_to_csv(list) do
    [["PREFIX", "ORIGINAL_URL", "NEW_URL"]] ++ list
      |> MyParser.dump_to_iodata()
      |> IO.iodata_to_binary
  end

  def add_attachment(binary_data) do
    %Bamboo.Attachment{content_type: "text/csv", filename: "data.csv", data: binary_data}
  end
Library Used:

:nimble_csv, bamboo_smtp

Can you help me how to make this process async.

This is what i want to do actually.

Post #5

NobbZ

Thats the final goal. I meant that single snippet. It starts a single Task and then eventually tries to “add” the result of the call to a map. Though you can not “add” to a map. You can only add to numbers, and the BEAM does only know about 2 types of numbers, floating point numbers and integers.

Post #6

sahilpaudel

Yes, It was wrong of me doing that, I was just trying to mimic what was there in the Task.async documentation.

But that didn’t work and I am totally clueless how to go about that using Task.async.

Can you please help me here?

Post #7

NobbZ

Task.async/1 will start the job in a new process.

Task.await/1 will wait up to 5 seconds and then either return the result or crash the current process.

So the following is pretty similar to spawning many workers and then joining them at the end:

email_addresses
|> Enum.map(fn email ->
  Task.async(fn -> do_async_task(csv_path, email))
end)
|> Enum.each(&Task.await/1)

Make sure you do not need more than 5 seconds to send each individual email though ;D

You can also increase the timeout by using Task.await/2 instead:

email_addresses
|> Enum.map(fn email ->
  Task.async(fn -> do_async_task(csv_path, email))
end)
|> Enum.each(&Task.await(&1, time_out))

Here you can see an async sleep in “action”:

iex(2)> :timer.tc(fn ->
...(2)>   Enum.map(1..10, fn t ->
...(2)>     Task.async(fn ->
...(2)>       Process.sleep(t * 1000)
...(2)>     end)
...(2)>   end) |> Enum.each(&Task.await(&1, :infinity))
...(2)> end)
{10002421, :ok}

Post #8

sahilpaudel

Thank you @NobbZ I got the gist of how Task.async and await is working here.

I am just starting on elixir and so I have a request how can I achieve below task.

When the request comes to

call(csv_path, email)

It returns %{response: %{message: "Request Initiated"}}

And carry on executing the below function

Can I do that?

Also I wan’t to spawn new worker of each request for this route.

Thanks.

Post #9

lucaong

If you want to execute the task in the background as a side effect, without waiting for the result, you can use either Task.start_link/{1,3} or Task.Supervisor.start_child/{3,5}.

Another common way to perform computation asynchronously in the background is to use a background job library like Oban.

Post #10

Load more posts (4 remaining)