|
| 1 | +# frozen_string_literal: true |
| 2 | + |
| 3 | +module Journaled |
| 4 | + module DeliveryAdapters |
| 5 | + # Default delivery adapter that uses ActiveJob |
| 6 | + # |
| 7 | + # This adapter enqueues events to Journaled::DeliveryJob which |
| 8 | + # sends them to Kinesis. This is the default behavior and maintains |
| 9 | + # backward compatibility with previous versions of the gem. |
| 10 | + class ActiveJobAdapter < Journaled::DeliveryAdapter |
| 11 | + # Delivers events by enqueueing them to Journaled::DeliveryJob |
| 12 | + # |
| 13 | + # @param events [Array] Array of journaled events to deliver |
| 14 | + # @param enqueue_opts [Hash] Options for ActiveJob (priority, queue, wait, wait_until, etc.) |
| 15 | + # @return [void] |
| 16 | + def self.deliver(events:, enqueue_opts:) |
| 17 | + Journaled::DeliveryJob.set(enqueue_opts).perform_later(*delivery_perform_args(events)) |
| 18 | + end |
| 19 | + |
| 20 | + # Serializes events into the format expected by DeliveryJob |
| 21 | + # |
| 22 | + # @param events [Array] Array of journaled events |
| 23 | + # @return [Array<Hash>] Array of serialized event hashes |
| 24 | + def self.delivery_perform_args(events) |
| 25 | + events.map do |event| |
| 26 | + { |
| 27 | + serialized_event: event.journaled_attributes.to_json, |
| 28 | + partition_key: event.journaled_partition_key, |
| 29 | + stream_name: event.journaled_stream_name, |
| 30 | + } |
| 31 | + end |
| 32 | + end |
| 33 | + |
| 34 | + # Returns the database connection to use for transactional batching |
| 35 | + # |
| 36 | + # This is determined by the configured queue adapter, since ActiveJob |
| 37 | + # enqueues jobs to the same database that should be used for transactions. |
| 38 | + # |
| 39 | + # @return [ActiveRecord::ConnectionAdapters::AbstractAdapter] The connection to use |
| 40 | + def self.transaction_connection |
| 41 | + queue_adapter = Journaled.queue_adapter |
| 42 | + |
| 43 | + if queue_adapter.in? %w(delayed delayed_job) |
| 44 | + Delayed::Job.connection |
| 45 | + elsif queue_adapter == 'good_job' |
| 46 | + GoodJob::BaseRecord.connection |
| 47 | + elsif queue_adapter == 'que' |
| 48 | + Que::ActiveRecord::Model.connection |
| 49 | + elsif queue_adapter == 'test' && Rails.env.test? |
| 50 | + ActiveRecord::Base.connection |
| 51 | + else |
| 52 | + raise "Unsupported queue adapter: #{queue_adapter}" |
| 53 | + end |
| 54 | + end |
| 55 | + |
| 56 | + # Validates that a supported queue adapter is configured |
| 57 | + # |
| 58 | + # @return [void] |
| 59 | + def self.validate_configuration! |
| 60 | + unless Journaled::SUPPORTED_QUEUE_ADAPTERS.include?(Journaled.queue_adapter) |
| 61 | + raise <<~MSG |
| 62 | + Journaled has detected an unsupported ActiveJob queue adapter: `:#{Journaled.queue_adapter}` |
| 63 | +
|
| 64 | + Journaled jobs must be enqueued transactionally to your primary database. |
| 65 | +
|
| 66 | + Please install the appropriate gems and set `queue_adapter` to one of the following: |
| 67 | + #{Journaled::SUPPORTED_QUEUE_ADAPTERS.map { |a| "- `:#{a}`" }.join("\n")} |
| 68 | +
|
| 69 | + Read more at https://github.com/Betterment/journaled |
| 70 | + MSG |
| 71 | + end |
| 72 | + end |
| 73 | + end |
| 74 | + end |
| 75 | +end |
0 commit comments