Class EventMachine::Iterator
In: lib/em/iterator.rb
Parent: Object

A simple iterator for concurrent asynchronous work.

Unlike ruby‘s built-in iterators, the end of the current iteration cycle is signaled manually, instead of happening automatically after the yielded block finishes executing. For example: 

  (0..10).each{ |num| }

becomes: 

  EM::Iterator.new(0..10).each{ |num,iter| iter.next }

This is especially useful when doing asynchronous work via reactor libraries and functions. For example, given a sync and async http api: 

  response = sync_http_get(url); ...
  async_http_get(url){ |response| ... }

a synchronous iterator such as: 

  responses = urls.map{ |url| sync_http_get(url) }
  ...
  puts 'all done!'

could be written as: 

  EM::Iterator.new(urls).map(proc{ |url,iter|
    async_http_get(url){ |res|
      iter.return(res)
    }
  }, proc{ |responses|
    ...
    puts 'all done!'
  })

Now, you can take advantage of the asynchronous api to issue requests in parallel. For example, to fetch 10 urls at a time, simply pass in a concurrency of 10: 

  EM::Iterator.new(urls, 10).each do |url,iter|
    async_http_get(url){ iter.next }
  end

Methods

concurrency=   each   inject   map   new  

Attributes

concurrency  [R] 

Public Class methods

new(list, concurrency = 1)

Create a new parallel async iterator with specified concurrency. 

  i = EM::Iterator.new(1..100, 10)

will create an iterator over the range that processes 10 items at a time. Iteration is started via each, map or inject

Public Instance methods

concurrency=(val)

Change the concurrency of this iterator. Workers will automatically be spawned or destroyed to accomodate the new concurrency level.

each(foreach=nil, after=nil, &blk)

Iterate over a set of items using the specified block or proc. 

  EM::Iterator.new(1..100).each do |num, iter|
    puts num
    iter.next
  end

An optional second proc is invoked after the iteration is complete. 

  EM::Iterator.new(1..100).each(
    proc{ |num,iter| iter.next },
    proc{ puts 'all done' }
  )
inject(obj, foreach, after)

Inject the results of an asynchronous iteration onto a given object. 

  EM::Iterator.new(%w[ pwd uptime uname date ], 2).inject({}, proc{ |hash,cmd,iter|
    EM.system(cmd){ |output,status|
      hash[cmd] = status.exitstatus == 0 ? output.strip : nil
      iter.return(hash)
    }
  }, proc{ |results|
    p results
  })
map(foreach, after)

Collect the results of an asynchronous iteration into an array. 

  EM::Iterator.new(%w[ pwd uptime uname date ], 2).map(proc{ |cmd,iter|
    EM.system(cmd){ |output,status|
      iter.return(output)
    }
  }, proc{ |results|
    p results
  })

[Validate]