DEV Community: Mykola Zozuliak

How to use ActionCable with async requests in a Ruby on Rails web app

Mykola Zozuliak — Fri, 17 May 2024 18:16:34 +0000

Problem

You have a complex Ruby on Rails web application. This application consists of 2 parts:

Front-end that renders HTML pages
The back-end serves the data to those web pages using AJAX requests.

There is one specific page that when a user open’s It — the front-end is making a request to a Ruby on Rails server and Ruby code needs to make a request to an external API to get the data requested. Pretty straightforward scenario in modern web applications. The issue is that the mentioned API call is very slow. Let’s say that It can take up to a minute to return a response. As I mentioned It is an external API, so we have no control over how well It performs and we can’t make any changes there. So how to solve that issue? Well, there are a few things we can do:

Straight forward approach. Every web server has a timeout configuration, so if your web server has a timeout set to 30 seconds, but requests take more time, a user will receive 504 error — https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/504. You can increase a timeout, but there is no guarantee, that the next time you will make a request It will again be longer than the set number. So this solution will not be stable.
Heat cache approach. We can heat the cache. If we know all the possible API requests we need to do during the day, we can schedule a worker that will put all those requests and their responses into the Rails cache with a 24-hour (or any other time a developer identifies as reasonable) expiration time. So every time a user opens a page, data will be taken directly from the cache instantly. This is a great solution, but It will only work if we know all possible API requests that any user can do and that this number is reasonable, It should not be let’s say 100K requests. You will need to have a huge cache that will be very expansive too. So this solution can only work in specific scenarios.
Async API requests inside workers. The third and last option that I see here requires both front-end and back-end work. The idea is simple — the front-end triggers a request, and controller action schedules a Sidekiq worker (ActiveJob job), a worker collects the response and puts it in the cache, then sends a message to the ActionCable channel, the front-end receives a message that data is ready, triggers another request to get data from the cache. Now let’s move to the next section for the detailed solution.

Solution

Steps that the application needs to follow:

Step 1. Front-end trigger request. I will not post any code here because any front-end framework can be used to do that. The idea is simple, frontend should send a request, saying that the data is needed. After that, we should have some designed UI state that will show some message to a user, like: We are collecting a data for you, just wait...
Step 2. The Rails server receives a request and schedules a worker. In order to make this correct, let’s create a simple Ruby service that will do all the needed work and then just call this service in the controller action:

class AsyncRequest
  def self.call(endpoint, params)
    hash = "async_request_#{endpoint}?#{params.to_query}"
    cached_response = Rails.cache.fetch(hash, expires_in: 2.hours)
    return { pending: false, hash: hash, data: cached_response } if cached_response

    AsyncQueryWorker.perform_async(endpoint, params, hash)
    { pending: true, hash: hash }
  end
end

The logic is very simple, if this request was already cached, get It from the cache and return right away. In the opposite scenario call the AsyncQueryWorker worker and return { pending: true, hash: hash } hash to the frontend. Frontend should remember the hash value. Controller action will look as simple as this:

class PagesController < ApplicationController
  def show
    respond_to do |format|
      format.json do
        render json: AsyncRequest.call('some_api/endpoint.json', permitted_params)
      end
    end
  end
end

Step 3. Frontend needs to subscribe to a WebSocket channel, in order to receive a notification that the worker has done processing the request. I will add some ReactJS code here (because this is the framework I am using), so you will get an idea of what needs to be done. PS: any front-end framework will do the trick too.

  const [websocketSub, setWebsocketSub] = useState(null);  
  const hash = ''; // a hash that front-end needs to record, received from backend in the previous step
  const token = ''; // a user token, that needs to be placed in meta tag. Read Rails guide to learn more
  useEffect(() => {
    if (!isUndefined(hash)) {
      if (token.length > 0 && !websocketSub) {
        App.initCable(token);
        const sub = App.cable.subscriptions.create({ channel: 'AsyncQueryChannel', hash: hash }, {
          received: (event) => {
            if (event.query_status === 'processed') {
              doRequest();
              // do the same request to the backend as above, because It's
              // params are converted to hash, It will be taken from the cache
            } else if (event.query_status === 'failed') {
              // display some UI error
            }
          },
        });
        setWebsocketSub(sub);
      }
      // initial request here
      doRequest();
    }

    return () => {
      if (App.cable) {
        each(App.cable.subscriptions.subscriptions, (sub) => {
          if (includes(sub.identifier, 'AsyncQueryChannel')) {
            App.cable.subscriptions.remove(sub);
          }
        });
        setWebsocketSub(null);
        // we need to remove subscription when request cycle is done
      }
    };
  }, [hash]);

Step 4. Now while the front is in loading state, it’s time to implement the worker itself.

class AsyncQueryWorker
  include Sidekiq::Worker

  sidekiq_options queue: :async_quaries, retry: 0 # add new separate queue
  # we can have retries here, but It will increase the amount of time
  # a user is waiting for a response, so let's not do this

  def perform(endpoint, params, hash)
    @endpoint = endpoint # a url where to make an API request
    @params = params # params for that request
    @hash = hash # uniq hash string that will be used for caching
    return if request_in_progress? # to be sure we will not process same request more the once

    do_request
  end

  private

  def do_request
    request
    $redis.redis.set(redis_key, @hash, ex: 300) # if something will stuck, in 300 seconds, redis will clear this request
    ActionCable.server.broadcast("async_query_#{@hash}", { query_status: 'processed' })
  rescue Faraday::Error
    ActionCable.server.broadcast("async_query_#{@hash}", { query_status: 'failed' })
    $redis.redis.del(redis_key)
  end

  def request_in_progress?
    request_in_progress = true
    $redis.redis.watch(redis_key) do
      request_in_progress = $redis.redis.get(redis_key).present?
      $redis.redis.unwatch
    end
    request_in_progress
  end

  def connection
    @_connection ||= Faraday.new(url: @endpoint) do |faraday|
      faraday.request :url_encoded
      faraday.headers['Content-Type'] = content_type
      faraday.options.timeout = 45
      faraday.response(:logger)
      faraday.adapter Faraday.default_adapter
    end
  end

  def request
    @_request ||= connection.get(@endpoint, @params)
  end

  def redis_key
    @_redis_key ||= "async-request-#{@hash}"
  end
end

We also need to create a channel class:

class AsyncQueryChannel < ApplicationCable::Channel
  def subscribed
    stream_from "async_query_#{params[:hash]}"
  end
end

Step 5. Frontend code receives an event when that data is ready or when an error occurs through the web socket. Done)

Summary

I believe this is a great solution for a scenario when you are using some not performance reliable external API that will be triggered by a web page. Why do I think so? Because I did in my app years ago and It works)

Copy of my original post: https://medium.com/@zozulyak.nick/how-to-use-actioncable-with-async-requests-in-a-ruby-on-rails-web-app-a70820c1fabe

Concurrent-ruby (async) S3 files download

Mykola Zozuliak — Fri, 17 May 2024 18:10:08 +0000

Problem

Let’s say we need to traverse through thousands of files in our S3 Storage in a Ruby app. Let’s say we have a bunch of logs there that we need to read every day and process. If we just use a straightforward approach, like opening, reading, and processing every file one by one, our solution will work, but It will take a lot of time to process. So we need to improve the speed. Here ruby-concurrent gem is our helper https://github.com/ruby-concurrency/concurrent-ruby. Why do we need this gem? Because It’s simpler to use than Threads and this gem has a lot more features in It. In this article, we will use Concurrent::Promises.future as the most common use of concurrent code. Because reading a file from S3 is an IO operation, we can get a huge benefit in speed if we gonna use concurrent code doing HTTP requests. Remember that concurrency will not give you speed improvements if in every Promise or Thread you will do any calculations. Because of Ruby GIL, every thread will be blocked until calculations are finished.

Solution

Step 1
Let’s start building our small project. First, let’s add needed gems:

gem 'concurrent-ruby'
gem 'aws-sdk-s3', '~> 1'
gem 'dotenv', groups: [:development, :test]
gem 'benchmark'

Summary of added gems:

concurrent-ruby — our main instrument to write concurrent code as described above
aws-sdk-s3 — official AWS S3 SDK gem to work with S3 Storage
dotenv — a great gem that helps you put our secret keys in .env file on your local machine. We need this to configure AWS SDK as described here https://github.com/aws/aws-sdk-ruby?tab=readme-ov-file#configuration benchmark — to measure the speed

Step 2
To read files from S3, let’s create a new Ruby class — S3Storage:

class S3Storage
  attr_reader :bucket_name

  def initialize(bucket_name)
    @bucket_name = bucket_name
  end

  def get_files(keys)
    keys.each_with_object([]) do |key, events|
      events << bucket.objects(prefix: key).to_a
    end.flatten
  end

  private

  def client
    @_client ||= Aws::S3::Client.new
  end

  def bucket
    @_bucket ||= Aws::S3::Bucket.new(bucket_name, client: client)
  end
end

Here we need to traverse through all files (objects) in needed folders (keys). This part is needed because we don’t know how many files are there, but don’t worry. With this approach, we will have an array of ObjectSummary objects. Basically, it’s a pointer to a file that contains file metadata, but It does not include file content.

Now let’s create a method to read files in a certain folder and then open the first 100 of them to measure a code performance without the use of concurrency:

def process
  s3_storage = S3Storage.new('bucket-name')
  files = s3_storage.get_files(['path/folder-one', 'path/folder-two'])
  Benchmark.measure do
    files.first(100).each { |file| file.object.get.body.read }
    # here we just open files, in real project you can parse them and process they way you need
  end
end

The results are (total is 7.2 seconds):

0.525316   0.096526   0.621842 (  7.241752)

So the next step is doing the same but with concurrency:

THREADS_COUNT = 5 # we are testing with 5 threads

def process_with_threads(objects)
  objects.each_slice(THREADS_COUNT) do |logs|
    futures = logs.each_with_index.map do |log, i|
      Concurrent::Promises.future do
        log.object.get.body.read
        # here we just open files, in real project you can parse them and process they way you need
      end
    end.compact
    # we are composing 5 promises into one and then reading value from every one
    Concurrent::Promises.zip(*futures).value! if futures
    # when you call value! method, it means you are accessing the data in a promise
  end
end

def process_concurrent
  s3_storage = S3Storage.new('bucket-name')
  files = s3_storage.get_files(['path/folder-one', 'path/folder-two'])
  Benchmark.measure do
    process_with_threads(files.first(100))
  end
end

The results are (total is 2.9 seconds):

 0.444163   0.055578   0.499741 (  2.970004)

As you can see, with this approach you can easily speed up the reading files procedure using concurrency. I have run the same tests, but for 1000 files and the difference is even bigger in this case.

No concurenncy: 5.388176   0.716480   6.104656 ( 89.667263)
Concurrency: 3.732760   0.476933   4.209693 ( 24.761635)

PS: in my testing, I am using small .txt files that have S3 access logs in them, so that is another reason why reading those files does not take a lot of time.

So, take advantage of concurrency in Ruby for the tasks that It feets the most — like IO (input-output). Examples of such operations are:

HTTP requests
DB requests
Reading a file

A copy of my original post: https://medium.com/@zozulyak.nick/concurrent-ruby-async-s3-files-download-85f46f298319

Ruby class pattern to work with API requests with built-in async approach

Mykola Zozuliak — Thu, 16 May 2024 21:14:14 +0000

Intro

I have been working with dozens of different Web APIs (mostly JSON) as a Ruby developer. Some of them can provide an official Ruby gem that you can install and use in order to interact with the API in your Ruby code. Others don't have anything like It at all and you need to implement the interaction with the API yourself. In this case, there are a lot of helpful API ruby gems that are easier to use than the default Ruby URI module https://github.com/ruby/uri. Let's list a few of them:

Faraday - https://github.com/lostisland/faraday. Personally, I use It, just because historically it was linked to every project I worked with. In addition, you can install the Faraday Curl gem https://github.com/mauricio/faraday_curl to get nice Curl requests in logs.
HTTPParty - https://github.com/jnunemaker/httparty. Heard a lot about It, but never used it.
RESTClient - https://github.com/rest-client/rest-client for folks who prefer Sinatra's actions syntax.
etc...

Those tools are just tools that you can use in order to implement the connection and interaction with any Web API. But the problem is that every API has Its interface, authorization rules, rate limiter, the way It handles errors, etc. So if in your project you need to connect 2 totally different APIs - you need to implement the interaction to each one yourself (assuming that those APIs have no gems available). If that API interaction code will not be properly encapsulated in Ruby classes, every developer in your team will need to carefully read API documentation in order to use the code that triggers those API calls. With every added API to your project, this issue will be a bigger and bigger problem with time. In this article, I will provide my own idea of how to create a simple, but powerful Ruby class pattern to hide HTTP interaction for any API in a Ruby class. This idea is based on the work that I did with my colleague years ago. Every such class will have the same external interface. As a bonus - my solution includes a way to do async HTTP requests.

Pattern

We should start with the list of Ruby gems that I am gonna use In my example:

gem 'concurrent-ruby'
gem 'faraday'
gem 'faraday_curl'

concurrent-ruby - to add async requests ability. https://github.com/ruby-concurrency/concurrent-ruby
faraday - to do HTTP requests (you can use any approach to do HTTP requests as I mentioned in the Intro section)
faraday_curl - to get nice Curl requests in logs (totally optional)

The next step is to create a module that I am gonna call FaradayConnector - this module will have all the needed logic in order to make any kind of HTTP requests to Web API (JSON by default). This module will be used to mix in other modules (API-specific ones). I will add more clarity on that later in the article:

require 'faraday'
require 'faraday_curl'
require 'concurrent'

module FaradayConnector
  class ServerError < RuntimeError
    attr_reader :response

    def initialize(response)
      @response = response
      super
    end
  end
  class TimeoutError < RuntimeError; end
  class ClientError < RuntimeError; end

  def request
    return @_request if defined?(@_request)

    # creating a Promise for async approach
    @_request = Concurrent::Promises.future { do_request }
  end

  def process
    return @_process if defined?(@_process)

    request
    @_process = do_process
  end

  def as_json(_options = {})
    process
  end

  protected

  def do_request
    # implement the real request in Child
  end

  def do_process
    # implement additional response decorations in Child
    request.value!
  end

  def url
    # must be added in Child
    raise 'Undefined url'
  end

  def auth
    # must be added in Child or use nil, if API has no Authorization
    raise 'Undefined auth'
  end

  def additional_headers
    {}
  end

  def content_type
    'application/json'
  end

  def request_type
    :url_encoded
  end

  def get(path, params = {})
    handle_request { connection.get(path, params) }
  end

  def post(path, body = {})
    formatted_body = json_content? ? body.to_json : body
    handle_request { connection.post(path, formatted_body) }
  end

  def delete(path, params = {})
    handle_request { connection.delete(path, params) }
  end

  def put(path, body = {})
    formatted_body = json_content? ? body.to_json : body
    handle_request { connection.put(path, formatted_body) }
  end

  def timeout
    45
  end

  def connection
    @connection ||= Faraday.new(url: url) do |faraday|
      faraday.request request_type
      faraday.headers['Authorization'] = auth if auth
      faraday.headers['Content-Type'] = content_type
      faraday.headers = faraday.headers.merge(additional_headers) if additional_headers
      faraday.options.timeout = timeout
      faraday.response(:logger)
      faraday.request :curl, Logger.new($stdout), :info
      faraday.adapter Faraday.default_adapter
    end
  end

  def handle_request
    response = handle_errors { yield }
    parse_response(response)
  end

  # just an easier way to handle HTTP errors
  def handle_errors
    response = yield
    e = if [502, 504].include?(response.status)
          TimeoutError.new(response)
        elsif [500, 503].include?(response.status)
          ServerError.new(response)
        elsif [400, 401, 404, 422].include?(response.status)
          ClientError.new(response)
        end
    return response unless e

    raise e
  end

  def parse_response(response)
    return {} unless response.body

    json_content? ? JSON.parse(response.body) : response
  rescue JSON::ParserError
    {}
  end

  def json_content?
    content_type == 'application/json'
  end
end

Now, assume we need to connect 2 different APIs (I just googled 2 public ones to show as an example):

OpenMeteo - https://open-meteo.com/ to get weather forecast
SportMonks - https://docs.sportmonks.com/football/api/ to get stats for Football teams

Every API has Its own Authorization rules, OpenMeteo has none, and SportMonks uses an Authorization token. We need to create 2 folders with modules called ApiConnector:

open_meteo/api_connector.rb

require './faraday_connector'

module OpenMeteo
  module ApiConnector
    include FaradayConnector

    def url
      'https://api.open-meteo.com'
    end

    def auth
      # no auth for this example
    end
  end
end

sport_monks/api_connector.rb

require './faraday_connector'

module SportMonks
  module ApiConnector
    include FaradayConnector

    def url
      'https://api.sportmonks.com'
    end

    def auth
      # a token goes here based on the API documentation
      # https://docs.sportmonks.com/football/welcome/authentication
      'some-token'
    end
  end
end

With these 2 easy steps, we created a codebase that can successfully authorize those 2 APIs we are using.
Time to create some HTTP requests. Let's start with OpenMeteo. On their home page, there are 2 examples of 2 different endpoints - forecast and historical. The next step is to create a separate class for every endpoint using the already created ApiConnector module.

forecast

module OpenMeteo
  class Forecast
    include ApiConnector

    attr_reader :latitude, :longitude

    def initialize(latitude, longitude)
      @latitude = latitude
      @longitude = longitude
    end

    def do_request
      get('v1/forecast',
        latitude: latitude,
        longitude: longitude
      )
    end

    def do_process
      request.value!
      # additional data manipulations goes here
    end
  end
end

historical

module OpenMeteo
  class Historical
    include ApiConnector

    attr_reader :latitude, :longitude, :start_date, :end_date

    def initialize(latitude, longitude, start_date, end_date)
      @latitude = latitude
      @longitude = longitude
      @start_date = start_date
      @end_date = end_date
    end

    def do_request
      get('v1/era5',
        latitude: latitude,
        longitude: longitude,
        start_date: start_date,
        end_date: end_date
      )
    end

    def do_process
      request.value!
      # additional data manipulations goes here
    end

    private

    def url
      # you can change url when needed easily
      'https://archive-api.open-meteo.com/'
    end
  end
end

There are 3 essential methods here:

initialize - a class constructor is only needed when an API request has input parameters
do_request - a place where you are triggering an HTTP request (available methods are get, post, put, delete). Remember this method returns Concurrent::Promise class instances.
do_process - calling value! method on Concurrent::Promise a class instance is essential to get the request-response.

Note: You probably noticed the url method override in the second file, we need this because a second API call has a different URL, and with this pattern, it's very easy to do)

This is the way to use those classes to do requests:

forecast = OpenMeteo::Forecast.new(42.361145, -71.057083)
historical = OpenMeteo::Historical.new(42.361145, -71.057083, '2024-01-01', '2024-01-30')
# Async part
forecast.request
historical.request

puts 'historical', historical.process
puts 'forecast', forecast.process

With the async approach, it means that we are doing both requests in 2 different threads, so they will work faster. When you call process method on any of It, promise will wait until all requests will be finished before returning any results, this is the way to synchronize code back.

The output will look like this:

I, [2024-05-09T14:34:30.756703 #74264]  INFO -- request: GET https://api.open-meteo.com/v1/forecast?latitude=42.361145&longitude=-71.057083
I, [2024-05-09T14:34:30.756857 #74264]  INFO -- request: Content-Type: "application/json"
User-Agent: "Faraday v2.9.0"
I, [2024-05-09T14:34:30.756822 #74264]  INFO -- request: GET https://archive-api.open-meteo.com/v1/era5?end_date=2024-01-30&latitude=42.361145&longitude=-71.057083&start_date=2024-01-01
I, [2024-05-09T14:34:30.761810 #74264]  INFO -- request: Content-Type: "application/json"
User-Agent: "Faraday v2.9.0"
I, [2024-05-09T14:34:31.084956 #74264]  INFO -- response: Status 200
I, [2024-05-09T14:34:31.085185 #74264]  INFO -- response: date: "Thu, 09 May 2024 18:34:31 GMT"
content-type: "application/json; charset=utf-8"
transfer-encoding: "chunked"
connection: "keep-alive"
content-encoding: "gzip"
I, [2024-05-09T14:34:31.090995 #74264]  INFO -- response: Status 200
I, [2024-05-09T14:34:31.091058 #74264]  INFO -- response: date: "Thu, 09 May 2024 18:34:31 GMT"
content-type: "application/json; charset=utf-8"
transfer-encoding: "chunked"
connection: "keep-alive"
content-encoding: "gzip"
historical
{"latitude"=>42.355007, "longitude"=>-71.12906, "generationtime_ms"=>0.0050067901611328125, "utc_offset_seconds"=>0, "timezone"=>"GMT", "timezone_abbreviation"=>"GMT", "elevation"=>9.0}
forecast
{"latitude"=>42.36515, "longitude"=>-71.0618, "generationtime_ms"=>0.00095367431640625, "utc_offset_seconds"=>0, "timezone"=>"GMT", "timezone_abbreviation"=>"GMT", "elevation"=>9.0}

As we can see, both requests are running at the same time, so we are sure the async approach is working. If we just run process methods for both instances one by one, we will have a synchronized approach, that will look like this:

forecast = OpenMeteo::Forecast.new(42.361145, -71.057083)
historical = OpenMeteo::Historical.new(42.361145, -71.057083, '2024-01-01', '2024-01-30')
puts 'historical', historical.process
puts 'forecast', forecast.process

I, [2024-05-09T14:37:15.830733 #74386]  INFO -- request: GET https://archive-api.open-meteo.com/v1/era5?end_date=2024-01-30&latitude=42.361145&longitude=-71.057083&start_date=2024-01-01
I, [2024-05-09T14:37:15.830765 #74386]  INFO -- request: Content-Type: "application/json"
User-Agent: "Faraday v2.9.0"
I, [2024-05-09T14:37:16.154027 #74386]  INFO -- response: Status 200
I, [2024-05-09T14:37:16.154220 #74386]  INFO -- response: date: "Thu, 09 May 2024 18:37:16 GMT"
content-type: "application/json; charset=utf-8"
transfer-encoding: "chunked"
connection: "keep-alive"
content-encoding: "gzip"
historical
{"latitude"=>42.355007, "longitude"=>-71.12906, "generationtime_ms"=>0.0050067901611328125, "utc_offset_seconds"=>0, "timezone"=>"GMT", "timezone_abbreviation"=>"GMT", "elevation"=>9.0}
I, [2024-05-09T14:37:16.157443 #74386]  INFO -- request: GET https://api.open-meteo.com/v1/forecast?latitude=42.361145&longitude=-71.057083
I, [2024-05-09T14:37:16.157511 #74386]  INFO -- request: Content-Type: "application/json"
User-Agent: "Faraday v2.9.0"
I, [2024-05-09T14:37:16.495799 #74386]  INFO -- response: Status 200
I, [2024-05-09T14:37:16.496003 #74386]  INFO -- response: date: "Thu, 09 May 2024 18:37:16 GMT"
content-type: "application/json; charset=utf-8"
transfer-encoding: "chunked"
connection: "keep-alive"
content-encoding: "gzip"
forecast
{"latitude"=>42.36515, "longitude"=>-71.0618, "generationtime_ms"=>0.00095367431640625, "utc_offset_seconds"=>0, "timezone"=>"GMT", "timezone_abbreviation"=>"GMT", "elevation"=>9.0}

Now let's get back to the second API that we planned to connect. It has a lot more examples that you can check out here https://docs.sportmonks.com/football/api/demo-response-files. To be very simple, let's say we want to get data about some team with an id 8.

module SportMonks
  class Team
    include ApiConnector

    attr_reader :team_id

    def initialize(team_id)
      @team_id = team_id
    end

    def do_request
      get("v3/football/teams/#{team_id}")
    end

    def do_process
      request.value!
      # additional data manipulations goes here
    end
  end
end

team = SportMonks::Team.new(8)
team.request
team.process

Same interface, but a different API.

Summary

Feel free to use this pattern for your API needs. I have been using this for years for a wide variety of services. I prefer using this, then connecting API-specific gems. From my perspective, this approach has a few benefits:

integrated async way of doing requests
same external interface for every API you will use in your codebase, which makes It very easy to use by any developer in your team
once ApiConnector is created with configured authorization, you will need to think about It ever again. (always use a secure way to store tokens or other secrets). If you don't know what to use, I can suggest dotenv gem - https://github.com/bkeepers/dotenv
flexible request configuration that is encapsulated in one class with a method override approach

PS: It's a copy of my original post on Medium - https://medium.com/@zozulyak.nick/ruby-class-pattern-to-work-with-api-requests-with-built-in-async-approach-bf0713a7dc96