Class: Radiator::Stream

Inherits:
Api
  • Object
show all
Defined in:
lib/radiator/stream.rb

Overview

Radiator::Stream allows a live view of the STEEM blockchain.

All values returned by ‘get_dynamic_global_properties` can be streamed.

For example, if you want to know which witness is currently signing blocks, use the following:

stream = Radiator::Stream.new
stream.current_witness do |witness|
  puts witness
end

Constant Summary collapse

INITIAL_TIMEOUT = 
0.0200
MAX_TIMEOUT = 
80
MAX_BLOCKS_PER_NODE = 
100

Instance Method Summary collapse

Methods inherited from Api

#api_name, #find_account, #find_block, #get_blocks, #respond_to_missing?, #steem_per_mvest, #steem_per_usd

Constructor Details

#initialize(options = {}) ⇒ Stream

Returns a new instance of Stream.



18
19
20
21
 
# File 'lib/radiator/stream.rb', line 18

def initialize(options = {})
  @api_options = options
  @logger = @api_options[:logger] || Radiator.logger
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(m, *args, &block) ⇒ Object 



200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/radiator/stream.rb', line 200

def method_missing(m, *args, &block)
  super unless respond_to_missing?(m)
  
  @latest_values ||= []
  @latest_values.shift(5) if @latest_values.size > 20
  loop do
    value = if (n = method_params(m)).nil?
      key_value = api.get_dynamic_global_properties.result[m]
    else
      key = n.keys.first
      if !!n[key]
        r = api.get_dynamic_global_properties.result
        key_value = param = r[n[key]]
        result = nil
        loop do
          response = api.send(key, param)
          raise StreamError, JSON[response.error] if !!response.error
          result = response.result
          break if !!result
          @logger.warn "#{key}: #{param} result missing, retrying with timeout: #{@timeout || INITIAL_TIMEOUT} seconds"
          shutdown
          sleep timeout
        end
        @timeout = INITIAL_TIMEOUT
        result
      else
        key_value = api.get_dynamic_global_properties.result[key]
      end
    end
    unless @latest_values.include? key_value
      @latest_values << key_value
      if !!block
        yield value
      else
        return value
      end
    end
    sleep 0.0200
  end
end

Instance Method Details

#api ⇒ Object 



23
24
25
 
# File 'lib/radiator/stream.rb', line 23

def api
  @api ||= Api.new(@api_options)
end

#blocks(start = nil, mode = :irreversible, max_blocks_per_node = MAX_BLOCKS_PER_NODE, &block) ⇒ Hash

Returns the latest blocks from the blockchain.

Parameters:

  • start (defaults to: nil) —

    starting block

  • mode (defaults to: :irreversible) —

    we have the choice between

    • :head the last block

    • :irreversible the block that is confirmed by 2/3 of all block producers and is thus irreversible!

    • :max_blocks_per_node the number of blocks to read before trying a new node

  • block —

    the block to execute for each result, optional.

Returns:

  • (Hash) 


157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/radiator/stream.rb', line 157

def blocks(start = nil, mode = :irreversible, max_blocks_per_node = MAX_BLOCKS_PER_NODE, &block)
  counter = 0
  
  if start.nil?
    properties = api.get_dynamic_global_properties.result
    start = case mode.to_sym
    when :head then properties.head_block_number
    when :irreversible then properties.last_irreversible_block_num
    else; raise StreamError, '"mode" has to be "head" or "irreversible"'
    end
  end
  
  loop do
    properties = api.get_dynamic_global_properties.result
    
    head_block = case mode.to_sym
    when :head then properties.head_block_number
    when :irreversible then properties.last_irreversible_block_num
    else; raise StreamError, '"mode" has to be "head" or "irreversible"'
    end
    
    [*(start..(head_block))].each do |n|
      if (counter += 1) > max_blocks_per_node
        shutdown
        counter = 0
      end

      response = api.send(:get_block, n)
      raise StreamError, JSON[response.error] if !!response.error
      result = response.result
    
      if !!block
        yield result
      else
        return result
      end
    end
    
    start = head_block + 1
    sleep 3
  end
end

#method_names ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/radiator/stream.rb', line 27

def method_names
  @method_names ||= [
    :head_block_number,
    :head_block_id,
    :time,
    :current_witness,
    :total_pow,
    :num_pow_witnesses,
    :virtual_supply,
    :current_supply,
    :confidential_supply,
    :current_sbd_supply,
    :confidential_sbd_supply,
    :total_vesting_fund_steem,
    :total_vesting_shares,
    :total_reward_fund_steem,
    :total_reward_shares2,
    :total_activity_fund_steem,
    :total_activity_fund_shares,
    :sbd_interest_rate,
    :average_block_size,
    :maximum_block_size,
    :current_aslot,
    :recent_slots_filled,
    :participation_count,
    :last_irreversible_block_num,
    :max_virtual_bandwidth,
    :current_reserve_ratio,
    :block_numbers,
    :blocks
  ].freeze
end

#method_params(method) ⇒ Object 



60
61
62
63
64
65
66
 
# File 'lib/radiator/stream.rb', line 60

def method_params(method)
  case method
  when :block_numbers then {head_block_number: nil}
  when :blocks then {get_block: :head_block_number}
  else; nil
  end
end

#operations(type = nil, start = nil, mode = :irreversible, &block) ⇒ Hash

Returns the latest operations from the blockchain.

If symbol are passed, then only that operation is returned. Expected symbols are:

transfer_to_vesting
withdraw_vesting
interest
transfer
liquidity_reward
author_reward
curation_reward
transfer_to_savings
transfer_from_savings
cancel_transfer_from_savings
escrow_transfer
escrow_approve
escrow_dispute
escrow_release
comment
limit_order_create
limit_order_cancel
fill_convert_request
fill_order
vote




witness_update
pow
custom

Parameters:

  • type (symbol || Array<symbol>) (defaults to: nil) —

    the type(s) of operation, optional.

  • start (defaults to: nil) —

    starting block

  • mode (defaults to: :irreversible) —

    we have the choice between

    • :head the last block

    • :irreversible the block that is confirmed by 2/3 of all block producers and is thus irreversible!

  • block —

    the block to execute for each result, optional.

Returns:

  • (Hash) 


108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/radiator/stream.rb', line 108

def operations(type = nil, start = nil, mode = :irreversible, &block)
  transactions(start, mode) do |transaction|
    ops = transaction.operations.map do |t, op|
      t = t.to_sym
      if type == t
        op
      elsif type.nil? || [type].flatten.include?(t)
        {t => op}
      end
    end.compact
    
    next if ops.none?
    
    return ops unless !!block
    
    ops.each do |op|
      yield op
    end
  end
end

#shutdown ⇒ Object

Stops the persistant http connections.



250
251
252
253
254
255
256
257
258
 
# File 'lib/radiator/stream.rb', line 250

def shutdown
  begin
    @api.shutdown
  rescue => e
    @logger.warn("Unable to shut down: #{e}")
  end
  
  @api = nil
end

#timeout ⇒ Object 



241
242
243
244
245
246
 
# File 'lib/radiator/stream.rb', line 241

def timeout
  @timeout ||= INITIAL_TIMEOUT
  @timeout *= 2
  @timeout = INITIAL_TIMEOUT if @timeout > MAX_TIMEOUT
  @timeout
end

#transactions(start = nil, mode = :irreversible, &block) ⇒ Hash

Returns the latest transactions from the blockchain.

Parameters:

  • start (defaults to: nil) —

    starting block

  • mode (defaults to: :irreversible) —

    we have the choice between

    • :head the last block

    • :irreversible the block that is confirmed by 2/3 of all block producers and is thus irreversible!

  • block —

    the block to execute for each result, optional.

Returns:

  • (Hash) 


137
138
139
140
141
142
143
144
145
146
 
# File 'lib/radiator/stream.rb', line 137

def transactions(start = nil, mode = :irreversible, &block)
  blocks(start, mode) do |b|
    next if (_transactions = b.transactions).nil?
    return _transactions unless !!block
    
    _transactions.each.each do |transaction|
      yield transaction
    end
  end
end