Inter-Process Communication With Ruby

2023-08-28

Inter-Process Communication With Ruby 🔄

In 2015, I needed a way to record audio streams on a scheduled basis. To accomplish this, I created a Rails app called Radioshifter that served as a UI for editing and storing schedules. In a separate process, a rufus-scheduler would run and start/stop various command line applications that captured online audio streams to disk and uploaded them to Dropbox.

Radioshifter - R.I.P. 🪦☠

Choosing a Method for Interprocess Communication

I looked at various methods for interprocess communication with Ruby, which included:

• pipes
• shared memory
• message queues
• sockets
• RPC frameworks
• Ruby dRb

dRb: The Right Tool for the Job

I ultimately chose dRb (distributed Ruby) for its simplicity and ease of use.

The dRb fronted process didn’t have too much to worry about in terms of processing messages. All it needed to do was take a message and edit rufus-scheduler’s job list. dRb seemed capable enough for that type of message load.

To enable communication between the Rails app and the scheduler, I used dRb. dRb is a distributed object system for Ruby. It allows you to expose objects to other processes and communicate with them. dRb is a part of the Ruby standard library, so it’s available out of the box. There are two main components to a dRb implementation, whatever you choose to be the server, and a client or clients.

My dRb Implementation in Radioshifter

Things to keep in mind:

• I did this in 2015, so some of the code may be a bit dated.
• I did this in a hurry, so there are some things that could be improved.
• At this point in time, there are probably better ways to do this
• There’s no built in or durability or retrying with dRb, like a robust message queue would have.
• I’ve never looked into the performace limits of dRb.

With that being said, there is a time and place for dRb, and it’s a good tool know about.

Below are the critical entry points for the dRb implementation, but you can also view the full application source code on the repository. I’m also no longer really fond of the -er -or naming convention that I used for certain classes, but I digress!

# The executable that starts the scheduler which is fronted by dRb

#!/usr/bin/env ruby
require 'drb'
require File.expand_path('../../config/environment', __FILE__ )

ENV['RAILS_ENV'] ||= 'development'
ENV['RUFUS_SCHEDULER_SERVER_PORT'] ||= '9000'
ENV['RECORDINGS_LOG_FOLDER'] ||= File.expand_path('../../log/recordings', __FILE__ )

trap("INT") { exit }

RUFUS_SCHEDULER = Rufus::Scheduler.new

server = Scheduler::Server.new(RUFUS_SCHEDULER)
bootstrapper = Scheduler::ServerBootstrapper.new(server, RecordingSchedule)
bootstrapper.bootstrap

DRb.start_service("druby://localhost:#{ENV['RUFUS_SCHEDULER_SERVER_PORT']}",
  server
)
DRb.thread.join  


# The server class that exposes the scheduler

module Scheduler
  class Server
    include ::Scheduler::JobTransformations

    def initialize(scheduler)
      @scheduler = scheduler
      @job_editor = Scheduler::JobEditor.new(@scheduler)
      @lock = Mutex.new
    end

    def cron(recording_schedule)
      @lock.synchronize do
        @job_editor.cron(recording_schedule)
      end
    end

    def at(recording_schedule)
      @lock.synchronize do
        @job_editor.at(recording_schedule)
      end
    end

    def unschedule(job_id)
      @lock.synchronize do
        @job_editor.unschedule(job_id)
      end
    end

    def job(job_id)
      begin
        job = @scheduler.job(job_id)
        job_to_hash(job) if job
      rescue ArgumentError => e
        nil
      end
    end

    def jobs(job_ids)
      found_jobs = job_ids.map do |id|
        job(id)
      end

      found_jobs.compact
    end
  end
end


# The dRb client initializer.
# In my app this is located at: config/initializers/rufus-scheduler.rb

require 'drb'
RUFUS_SCHEDULER_CLIENT = DRbObject.new(nil, 'druby://localhost:9000')

# Note: this must be called at least once per process to take any effect.
# This is particularly important if your application forks.
DRb.start_service


# Calling the dRb Client.

class RecordingSchedulesController < ApplicationController
  before_action :authenticate_user!
  before_action :redirect_to_time_zone, if: Proc.new { current_user.time_zone.blank? }

  def index
    @recording_schedules = RecordingSchedule.where(user_id: current_user.id).order(:name).decorate
  end

  def new
    @recording_schedule = RecordingSchedule.new
  end

  def create
    @recording_schedule = RecordingSchedule.new(recording_schedule_params).decorate
    @recording_schedule.user = current_user
    notifier = Services::ScheduleNotifier.new(@recording_schedule, RUFUS_SCHEDULER_CLIENT)

    if notifier.save
      redirect_to recording_schedules_path, flash: { success: 'Recording schedule added.' }
    else
      flash.now[:error] = "There were some problems with this recording schedule."
      render :new
    end
  end

  def destroy
    @recording_schedule = RecordingSchedule.find(params[:id]).decorate
    notifier = Services::ScheduleNotifier.new(@recording_schedule, RUFUS_SCHEDULER_CLIENT)
    flash_options = {}

    if notifier.destroy
      flash_options[:success] = 'Recording schedule removed.'
    else
      flash_options[:success] = 'Recording schedule not removed.'
    end

    redirect_to recording_schedules_path, flash_options
  end

  def edit
    @recording_schedule = RecordingSchedule.where(id: params[:id], user_id: current_user.id).first
  end

  def update
    @recording_schedule = RecordingSchedule.where(id: params[:id], user_id: current_user.id)
      .first
      .decorate
    notifier = Services::ScheduleNotifier.new(@recording_schedule, RUFUS_SCHEDULER_CLIENT)

    if notifier.save(recording_schedule_params)
      redirect_to recording_schedules_path, flash: { success: 'Changes saved.' }
    else
      flash.now[:error] = "There were some problems with this recording schedule."
      render :edit
    end
  end

  private

  def recording_schedule_params
    strip_blank_from_days_of_week(
      params
        .require(:recording_schedule)
        .permit(:duration, :stop_time, :name, :recurring,
         :start_on, :start_time, :stream_url, days_of_week: []
        )
    )
  end

  def redirect_to_time_zone
    redirect_to time_zone_path
  end

  def strip_blank_from_days_of_week(permittied_params)
    if permittied_params[:days_of_week].present?
      permittied_params[:days_of_week]
        .reject! {|i|i.blank?}
        .collect!(&:to_i)
    end
    permittied_params
  end
end

Conclusion

Overall, using dRb for interprocess communication in Ruby proved to be a reliable and efficient solution for my audio recording system. While there are other methods available for interprocess communication in Ruby, dRb’s simplicity and ease of use made it a good choice for my use case.