Class: ActiveRecord::FutureResult

Inherits:
Object
  • Object
show all
Defined in:
lib/active_record/future_result.rb

Overview

:nodoc:

Direct Known Subclasses

SelectAll

Defined Under Namespace

Classes: Complete, EventBuffer, SelectAll

Constant Summary collapse

Canceled = 
Class.new(ActiveRecordError)

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(pool, *args, **kwargs) ⇒ FutureResult

Returns a new instance of FutureResult.



64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/active_record/future_result.rb', line 64

def initialize(pool, *args, **kwargs)
  @mutex = Mutex.new

  @session = nil
  @pool = pool
  @args = args
  @kwargs = kwargs

  @pending = true
  @error = nil
  @result = nil
  @instrumenter = ActiveSupport::Notifications.instrumenter
  @event_buffer = nil
end

Instance Attribute Details

#lock_wait ⇒ Object (readonly)

Returns the value of attribute lock_wait.



62
63
64
 
# File 'lib/active_record/future_result.rb', line 62

def lock_wait
  @lock_wait
end

Class Method Details

.wrap(result) ⇒ Object 



50
51
52
53
54
55
56
57
 
# File 'lib/active_record/future_result.rb', line 50

def self.wrap(result)
  case result
  when self, Complete
    result
  else
    Complete.new(result)
  end
end

Instance Method Details

#cancel ⇒ Object 



92
93
94
95
96
 
# File 'lib/active_record/future_result.rb', line 92

def cancel
  @pending = false
  @error = Canceled
  self
end

#canceled? ⇒ Boolean

Returns:

  • (Boolean) 


133
134
135
 
# File 'lib/active_record/future_result.rb', line 133

def canceled?
  @session && !@session.active?
end

#execute!(connection) ⇒ Object 



88
89
90
 
# File 'lib/active_record/future_result.rb', line 88

def execute!(connection)
  execute_query(connection)
end

#execute_or_skip ⇒ Object 



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/active_record/future_result.rb', line 98

def execute_or_skip
  return unless pending?

  @pool.with_connection do |connection|
    return unless @mutex.try_lock
    begin
      if pending?
        @event_buffer = EventBuffer.new(self, @instrumenter)
        connection.with_instrumenter(@event_buffer) do
          execute_query(connection, async: true)
        end
      end
    ensure
      @mutex.unlock
    end
  end
end

#pending? ⇒ Boolean

Returns:

  • (Boolean) 


129
130
131
 
# File 'lib/active_record/future_result.rb', line 129

def pending?
  @pending && (!@session || @session.active?)
end

#result ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/active_record/future_result.rb', line 116

def result
  execute_or_wait
  @event_buffer&.flush

  if canceled?
    raise Canceled
  elsif @error
    raise @error
  else
    @result
  end
end

#schedule!(session) ⇒ Object 



83
84
85
86
 
# File 'lib/active_record/future_result.rb', line 83

def schedule!(session)
  @session = session
  @pool.schedule_query(self)
end

#then(&block) ⇒ Object 



79
80
81
 
# File 'lib/active_record/future_result.rb', line 79

def then(&block)
  Promise.new(self, block)
end