Class: ActiveRecord::Write
- Inherits:
-
Object
- Object
- ActiveRecord::Write
- Defined in:
- lib/active_record/write.rb,
lib/active_record/write/version.rb
Constant Summary collapse
- DEFAULT_SIZE =
24
- DEFAULT_SERIALIZER =
::JSON
- EMPTY_HASH =
{}
- VERSION =
"1.1.2"
Instance Method Summary collapse
-
#initialize(query:, columns:, target:, size:, serializer:, &transaction) ⇒ Write
constructor
‘query` is either an ActiveRecord query object or arel `columns` is a list of columns you want to have during the transaction `target` is the table you want to talk to `size` is the maximum number of running iterations in the pool, default: 24 `serializer` is the #dump duck for Array & Hash values, default: JSON `transaction` is the process you want to run against your database.
- #pool(qutex = Mutex.new) ⇒ Object
Constructor Details
#initialize(query:, columns:, target:, size:, serializer:, &transaction) ⇒ Write
‘query` is either an ActiveRecord query object or arel `columns` is a list of columns you want to have during the transaction `target` is the table you want to talk to `size` is the maximum number of running iterations in the pool, default: 24 `serializer` is the #dump duck for Array & Hash values, default: JSON `transaction` is the process you want to run against your database
21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 |
# File 'lib/active_record/write.rb', line 21 def initialize(query:, columns:, target:, size:, serializer:, &transaction) @query = query @columns = columns @target = target @size = size @serializer = serializer @transaction = transaction @table = Arel::Table.new(@target) @queue = case when activerecord? @query.pluck(*@columns) when arel? ActiveRecord::Base.connection.execute(@query.to_sql).map(&:values) when tuple? @query.map { |result| result.slice(*@columns).values } when twodimensional? @query else raise ArgumentError, 'query wasn\'t recognizable, please use some that looks like a: ActiveRecord::Base, Arel::SelectManager, Array<*Hash>, Array<*Array>' end puts "Migrating #{@queue.count} #{@target} records" end |
Instance Method Details
#pool(qutex = Mutex.new) ⇒ Object
44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/active_record/write.rb', line 44 def pool(qutex = Mutex.new) # Spin up a number of threads based on the `maximum` given 1.upto(@size).map do Thread.new do loop do # Try to get a new queue item item = qutex.synchronize { @queue.shift } if item.nil? # There is no more work break else # Wait for a free connection ActiveRecord::Base.connection_pool.with_connection do ActiveRecord::Base.transaction do # Execute each statement coming back Array[instance_exec(*item, &@transaction)].each do |instruction| ActiveRecord::Base.connection.execute(instruction.to_sql) end end end end end end end.map(&:join) end |