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

More importantly, full blocks, transactions, and operations can be streamed.

Constant Summary collapse

INITIAL_TIMEOUT = 
0.0200
MAX_TIMEOUT = 
80
MAX_BLOCKS_PER_NODE = 
1000
RANGE_BEHIND_WARNING = 
400

Constants inherited from Api

Api::DEFAULT_GOLOS_FAILOVER_URLS, Api::DEFAULT_GOLOS_URL, Api::DEFAULT_STEEM_FAILOVER_URLS, Api::DEFAULT_STEEM_URL, Api::HEALTH_URI, Api::POST_HEADERS

Instance Method Summary collapse

Methods inherited from Api

#api_name, #base_per_debt, #base_per_mvest, default_failover_urls, default_url, #find_account, #find_block, #get_blocks, #inspect, #respond_to_missing?, #stopped?

Methods included from Utils

#debug, #error, #extract_signatures, #hexlify, #pakArr, #pakC, #pakHash, #pakI, #pakL!, #pakS, #pakStr, #pakc, #paks, #send_log, #unhexlify, #varint, #warning

Constructor Details

#initialize(options = {}) ⇒ Stream

Returns a new instance of Stream.



28
29
30
 
# File 'lib/radiator/stream.rb', line 28

def initialize(options = {})
  super
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

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



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
 
# File 'lib/radiator/stream.rb', line 379

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
    break if stop?
    
    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
          break if stop?
          
          response = api.send(key, param)
          raise StreamError, JSON[response.error] if !!response.error
          result = response.result
          break if !!result
          warnning "#{key}: #{param} result missing, retrying with timeout: #{@timeout || INITIAL_TIMEOUT} seconds"
          reset_api
          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

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

Returns the latest blocks from the blockchain.

stream = Radiator::Stream.new
stream.blocks do |bk, num|
  puts "[#{num}] #{bk.to_json}"
end

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 (defaults to: 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) 


220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/radiator/stream.rb', line 220

def blocks(start = nil, mode = :irreversible, max_blocks_per_node = MAX_BLOCKS_PER_NODE, &block)
  reset_api
  
  replay = !!start
  counter = 0
  latest_block_number = -1
  @api_options[:max_requests] = [max_blocks_per_node * 2, @api_options[:max_requests].to_i].max
  
  loop do
    break if stop?
    
    catch :sequence do; begin
      head_block = api.get_dynamic_global_properties do |properties|
        break if stop?
        
        if properties.head_block_number.nil?
          # This can happen if a reverse proxy is acting up.
          standby "Bad block sequence after height: #{latest_block_number}", {
            and: {throw: :sequence}
          }
        end
            
        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
        
      if head_block == latest_block_number
        # This can when there's a delay in block production.
        sleep 0.5
        throw :sequence
      elsif head_block < latest_block_number
        # This can happen if a reverse proxy is acting up.
        standby "Invalid block sequence at height: #{head_block}", {
          and: {backoff: api, throw: :sequence}
        }
      end
      
      start ||= head_block
      range = (start..head_block)
      
      for n in range
        break if stop?

        if (counter += 1) > max_blocks_per_node
          reset_api
          counter = 0
        end
        
        if !replay && range.size > RANGE_BEHIND_WARNING
          # When the range is above RANGE_BEHIND_WARNING blocks, it's time
          # to warn, unless we're replaying.
          
          r = [*range]
          index = r.index(n)
          current_range = r[index..-1]
          
          if current_range.size % RANGE_BEHIND_WARNING == 0
            standby "Stream behind by #{current_range.size} blocks (about #{(current_range.size * 3) / 60.0} minutes)."
          end
        end
        
        block_api.get_block(n) do |current_block, error|
          if current_block.nil?
            standby "Node responded with: empty block, retrying ...", {
              and: {throw: :sequence}
            }
          elsif !!error
            standby "Node responded with: #{error.message || 'unknown error'}, retrying ...", {
              error: error,
              and: {throw: :sequence}
            }
          end
          
          latest_block_number = n
          return current_block, n if block.nil?
          yield current_block, n
        end
        
        start = head_block + 1
        sleep 3 / range.size
      end
    rescue StreamError; raise
    # rescue => e
    #   warning "Unknown streaming error: #{e.inspect}, retrying ...  "
    #   warning e
    #   redo
    end; end
  end
end

#method_names ⇒ Object 



337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
 
# File 'lib/radiator/stream.rb', line 337

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 



371
372
373
374
375
376
377
 
# File 'lib/radiator/stream.rb', line 371

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, options = {include_virtual: false}, &block) ⇒ Hash

Returns the latest operations from the blockchain.

stream = Radiator::Stream.new
stream.operations do |op|
  puts op.to_json
end

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






cancel_transfer_from_savings

claim_reward_balance
comment
comment_options
convert
custom
custom_json
decline_voting_rights
delegate_vesting_shares
delete_comment
escrow_approve
escrow_dispute
escrow_release
escrow_transfer
feed_publish
limit_order_cancel
limit_order_create
limit_order_create2
pow
pow2


set_withdraw_vesting_route
transfer
transfer_from_savings
transfer_to_savings
transfer_to_vesting
vote
withdraw_vesting
witness_update

For example, to stream only votes:

stream = Radiator::Stream.new
stream.operations(:vote) do |vote|
  puts vote.to_json
end

You can also stream virtual operations:

stream = Radiator::Stream.new
stream.operations(:author_reward) do |vop|
    puts "#{vop.author} got paid for #{vop.permlink}: #{[vop.sbd_payout, vop.steem_payout, vop.vesting_payout]}"
end

… or multiple virtual operation types;

stream = Radiator::Stream.new
stream.operations([:producer_reward, :author_reward]) do |vop|
  puts vop
end

… or all types, inluding virtual operation types;

stream = Radiator::Stream.new
stream.operations(nil, nil, :head, include_virtual: true) do |vop|
  puts vop
end

Expected virtual operation types:

producer_reward
author_reward
curation_reward
fill_convert_request
fill_order
fill_vesting_withdraw
interest
shutdown_witness

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.

  • options (Hash) (defaults to: {include_virtual: false}) —

    additional options

Options Hash (options):

  • :include_virtual (Boollean) —

    Also stream virtual options. Setting this true will impact performance. Default: false.

Returns:

  • (Hash) 


127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/radiator/stream.rb', line 127

def operations(type = nil, start = nil, mode = :irreversible, options = {include_virtual: false}, &block)
  type = [type].flatten.compact.map(&:to_sym)
  include_virtual = !!options[:include_virtual]
  
  if virtual_op_type?(type)
    include_virtual = true
  end
  
  latest_block_number = -1
  
  transactions(start, mode) do |transaction, trx_id, block_number|
    virtual_ops_collected = latest_block_number == block_number
    latest_block_number = block_number
    
    ops = transaction.operations.map do |t, op|
      t = t.to_sym
      if type.size == 1 && type.first == t
        op
      elsif type.none? || type.include?(t)
        {t => op}
      end
    end.compact
    
    if include_virtual && !virtual_ops_collected
      api.get_ops_in_block(block_number, true) do |vops|
        vops.each do |vtx|
          next unless defined? vtx.op
          
          t = vtx.op.first.to_sym
          op = vtx.op.last
          if type.size == 1 && type.first == t
            ops << op
          elsif type.none? || type.include?(t)
            ops << {t => op}
          end
        end
      end
      
      virtual_ops_collected = true
    end
    
    next if ops.none?
    
    return ops unless !!block
    
    ops.each do |op|
      yield op, trx_id, block_number
    end
  end
end

#shutdown ⇒ Object

Stops the persistant http connections.



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
 
# File 'lib/radiator/stream.rb', line 315

def shutdown
  flappy = false
  
  begin
    unless @api.nil?
      flappy = @api.send(:flappy?)
      @api.shutdown
    end
    
    unless @block_api.nil?
      flappy = @block_api.send(:flappy?) unless flappy
      @block_api.shutdown
    end
  rescue => e
    warning("Unable to shut down: #{e}")
  end
  
  @api = nil
  @block_api = nil
end

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

Returns the latest transactions from the blockchain.

stream = Radiator::Stream.new
stream.transactions do |tx, trx_id|
  puts "[#{trx_id}] #{tx.to_json}"
end

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) 


191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/radiator/stream.rb', line 191

def transactions(start = nil, mode = :irreversible, &block)
  blocks(start, mode) do |b, block_number|
    next if (_transactions = b.transactions).nil?
    return _transactions unless !!block
    
    _transactions.each_with_index do |transaction, index|
      trx_id = if !!b['transaction_ids']
        b['transaction_ids'][index]
      end
      
      yield transaction, trx_id, block_number
    end
  end
end