mongo_async.rb

require 'rx'
require 'mongo'

aggregate_query = [
    {
        '$group' => {
            '_id' => "$_id", 
            'count' => {
                '$sum' => 1
                }
            }
        },
    {
    	'$limit' => 5
    }
]

context  = {}

client = Mongo::Client.new([ '127.0.0.1:27017' ], :database => 'test')
collection = client['test']
find_patient_subject = Rx::Observable.start(
	lambda {
				collection.find({}, { projection: {name:1}}).limit(5)
				},
    context, 
    Rx::DefaultScheduler.instance
    )

count_patient_subject = Rx::Observable.start(
	lambda { 
    			collection.aggregate(aggregate_query)
				},
    context, 
    Rx::DefaultScheduler.instance
    )

count_subscription = count_patient_subject.subscribe(
    lambda {|x| x.each do |m| p m end},
    lambda {|e| p e},
    lambda {p 'aggregate completed'}
    )

patient_subscription = find_patient_subject.subscribe(
    lambda {|x| x.each do |m| p m end},
    lambda {|e| p e},
    lambda {p 'find completed'}
    )

sleep(20)