github-migration
/
mongo-ruby-driver
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Removed GetMoreMessage and QueryMessage classes with initial refactoring.

This commit is contained in:
Kyle Banker 2009-10-20 15:44:46 -04:00
parent 5586d53c6a
commit b938765ac0
9 changed files with 85 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lib/mongo.rb
View File

@ -3,6 +3,8 @@ require 'mongo/types/dbref'
require 'mongo/types/objectid' require 'mongo/types/objectid'
require 'mongo/types/regexp_of_holding' require 'mongo/types/regexp_of_holding'
require 'mongo/util/conversions'
require 'mongo/errors' require 'mongo/errors'
require 'mongo/connection' require 'mongo/connection'
require 'mongo/message' require 'mongo/message'

63
lib/mongo/cursor.rb
View File

@ -20,6 +20,7 @@ module Mongo
# A cursor over query results. Returned objects are hashes. # A cursor over query results. Returned objects are hashes.
class Cursor class Cursor
include Mongo::Conversions
include Enumerable include Enumerable
@ -246,11 +247,22 @@ module Mongo
end end
def refill_via_get_more def refill_via_get_more
if send_query_if_needed or @cursor_id == 0 return if send_query_if_needed || @cursor_id.zero?
return
end
@db._synchronize { @db._synchronize {
@db.send_to_db(GetMoreMessage.new(@admin ? 'admin' : @db.name, @collection.name, @cursor_id)) message = ByteBuffer.new
# Reserved.
message.put_int(0)
# DB name.
db_name = @admin ? 'admin' : @db.name
BSON.serialize_cstr(message, "#{db_name}.#{@collection.name}")
# Number of results to return; db decides for now.
message.put_int(0)
# Cursor id.
message.put_long(@cursor_id)
@db.send_message_with_operation_without_synchronize(OP_GET_MORE, message)
read_all read_all
} }
end end
@ -272,14 +284,55 @@ module Mongo
false false
else else
@db._synchronize { @db._synchronize {
@db.send_query_message(QueryMessage.new(@admin ? 'admin' : @db.name, @collection.name, @query)) message = construct_query_message(@query)
@query_run = true @query_run = true
@db.send_message_with_operation_without_synchronize(OP_QUERY, message)
read_all read_all
} }
true true
end end
end end
def construct_query_message(query)
message = ByteBuffer.new
message.put_int(query.query_opts)
db_name = @admin ? 'admin' : @db.name
BSON.serialize_cstr(message, "#{db_name}.#{@collection.name}")
message.put_int(query.number_to_skip)
message.put_int(query.number_to_return)
sel = query.selector
if query.contains_special_fields
sel = add_special_query_fields(sel, query)
end
message.put_array(BSON.new.serialize(sel).to_a)
message.put_array(BSON.new.serialize(query.fields).to_a) if query.fields
message
end
def add_special_query_fields(sel, query)
sel = OrderedHash.new
sel['query'] = query.selector
order_by = query.order_by
sel['orderby'] = get_query_order_by(order_by) if order_by
sel['$hint'] = query.hint if query.hint && query.hint.length > 0
sel['$explain'] = true if query.explain
sel['$snapshot'] = true if query.snapshot
sel
end
def get_query_order_by(order_by)
case order_by
when String then string_as_sort_parameters(order_by)
when Symbol then symbol_as_sort_parameters(order_by)
when Array then array_as_sort_parameters(order_by)
when Hash # Should be an ordered hash, but this message doesn't care
warn_if_deprecated(order_by)
order_by
else
raise InvalidSortValueError, "Illegal order_by, '#{query.order_by.class.name}'; must be String, Array, Hash, or OrderedHash"
end
end
def to_s def to_s
"DBResponse(flags=#@result_flags, cursor_id=#@cursor_id, start=#@starting_from)" "DBResponse(flags=#@result_flags, cursor_id=#@cursor_id, start=#@starting_from)"
end end

27
lib/mongo/db.rb
View File

@ -372,11 +372,6 @@ module Mongo
Cursor.new(self, collection, query, admin) Cursor.new(self, collection, query, admin)
end end
# Used by a Cursor to lazily send the query to the database.
def send_query_message(query_message)
send_to_db(query_message)
end
# Dereference a DBRef, getting the document it points to. # Dereference a DBRef, getting the document it points to.
def dereference(dbref) def dereference(dbref)
collection(dbref.namespace).find_one("_id" => dbref.object_id) collection(dbref.namespace).find_one("_id" => dbref.object_id)
@ -460,7 +455,7 @@ module Mongo
# Takes a MongoDB opcode, +operation+, and a message of class ByteBuffer, # Takes a MongoDB opcode, +operation+, and a message of class ByteBuffer,
# +message+, and sends the message to the databse, adding the necessary headers. # +message+, and sends the message to the databse, adding the necessary headers.
def send_message_with_operation(operation, message) def send_message_with_operation(operation, message)
_synchronize do @semaphore.synchronize do
connect_to_master if !connected? && @auto_reconnect connect_to_master if !connected? && @auto_reconnect
begin begin
message_with_headers = add_message_headers(operation, message) message_with_headers = add_message_headers(operation, message)
@ -474,6 +469,26 @@ module Mongo
end end
end end
def send_message_with_operation_without_synchronize(operation, message)
connect_to_master if !connected? && @auto_reconnect
begin
message_with_headers = add_message_headers(operation, message)
@logger.debug(" MONGODB #{operation} #{message}") if @logger
@socket.print(message_with_headers.to_s)
@socket.flush
rescue => ex
close
raise ex
end
end
def receive_message_with_operation(operation, message)
@semaphore.synchronize do
end
end
# Return +true+ if +doc+ contains an 'ok' field with the value 1. # Return +true+ if +doc+ contains an 'ok' field with the value 1.
def ok?(doc) def ok?(doc)
ok = doc['ok'] ok = doc['ok']

2
lib/mongo/errors.rb
View File

@ -31,5 +31,5 @@ module Mongo
class InvalidName < RuntimeError; end class InvalidName < RuntimeError; end
# Raised when the client supplies an invalid value to sort by. # Raised when the client supplies an invalid value to sort by.
class InvalidSortValueError < RuntimeError; end class InvalidSortValueError < MongoRubyError; end
end end

4
lib/mongo/message.rb
View File

@ -14,7 +14,7 @@
# limitations under the License. # limitations under the License.
# ++ # ++
%w(get_more_message kill_cursors_message message_header %w(kill_cursors_message message_header
msg_message query_message).each { |f| msg_message).each { |f|
require "mongo/message/#{f}" require "mongo/message/#{f}"
} }

32
lib/mongo/message/get_more_message.rb
View File

@ -1,32 +0,0 @@
# --
# Copyright (C) 2008-2009 10gen Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ++
require 'mongo/message/message'
require 'mongo/message/opcodes'
module Mongo
class GetMoreMessage < Message
def initialize(db_name, collection_name, cursor)
super(OP_GET_MORE)
write_int(0)
write_string("#{db_name}.#{collection_name}")
write_int(0) # num to return; leave it up to the db for now
write_long(cursor)
end
end
end

69
lib/mongo/message/query_message.rb
View File

@ -1,69 +0,0 @@
# --
# Copyright (C) 2008-2009 10gen Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ++
require 'mongo/message/message'
require 'mongo/message/opcodes'
require 'mongo/util/conversions'
require 'mongo/util/ordered_hash'
module Mongo
class QueryMessage < Message
include Mongo::Conversions
attr_reader :query
def initialize(db_name, collection_name, query)
super(OP_QUERY)
@query = query
@collection_name = collection_name
write_int(@query.query_opts)
write_string("#{db_name}.#{collection_name}")
write_int(query.number_to_skip)
write_int(query.number_to_return)
sel = query.selector
if query.contains_special_fields
sel = OrderedHash.new
sel['query'] = query.selector
if query.order_by
order_by = query.order_by
sel['orderby'] = case order_by
when String then string_as_sort_parameters(order_by)
when Symbol then symbol_as_sort_parameters(order_by)
when Array then array_as_sort_parameters(order_by)
when Hash # Should be an ordered hash, but this message doesn't care
warn_if_deprecated(order_by)
order_by
else
raise InvalidSortValueError.new("illegal order_by: is a #{query.order_by.class.name}, must be String, Array, Hash, or OrderedHash")
end
end
sel['$hint'] = query.hint if query.hint && query.hint.length > 0
sel['$explain'] = true if query.explain
sel['$snapshot'] = true if query.snapshot
end
write_doc(sel)
write_doc(query.fields) if query.fields
end
def first_key(key)
@first_key = key
end
def to_s
"db.#{@collection_name}.#{@query}"
end
end
end

58
test/messages/test_query_message.rb
View File

@ -1,58 +0,0 @@
# --
# Copyright (C) 2008-2009 10gen Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ++
$LOAD_PATH[0,0] = File.join(File.dirname(__FILE__), '..', 'lib')
require 'mongo'
require 'test/unit'
class TestQueryMessage < Test::Unit::TestCase
include Mongo
def test_timeout_opcodes
@timeout = true
@query = Query.new({}, nil, 0, 0, nil, nil, nil, @timeout)
@query_message = QueryMessage.new('db', 'collection', @query)
buf = @query_message.buf.instance_variable_get(:@buf)
assert_equal 0, buf[16]
@timeout = false
@query = Query.new({}, nil, 0, 0, nil, nil, nil, @timeout)
@query_message = QueryMessage.new('db', 'collection', @query)
buf = @query_message.buf.instance_variable_get(:@buf)
assert_equal 16, buf[16]
end
def test_timeout_opcodes
@timeout = true
@slave_ok = true
@query = Query.new({}, nil, 0, 0, nil, nil, nil, @timeout, @slave_ok)
@query_message = QueryMessage.new('db', 'collection', @query)
buf = @query_message.buf.instance_variable_get(:@buf)
assert_equal 4, buf[16]
@timeout = false
@slave_ok = true
@query = Query.new({}, nil, 0, 0, nil, nil, nil, @timeout, @slave_ok)
@query_message = QueryMessage.new('db', 'collection', @query)
buf = @query_message.buf.instance_variable_get(:@buf)
assert_equal 20, buf[16]
end
end

4
test/test_connection.rb
View File

@ -61,10 +61,8 @@ class TestConnection < Test::Unit::TestCase
logger = Logger.new(output) logger = Logger.new(output)
logger.level = Logger::DEBUG logger.level = Logger::DEBUG
db = Connection.new(@host, @port, :logger => logger).db('ruby-mongo-test') db = Connection.new(@host, @port, :logger => logger).db('ruby-mongo-test')
db['test'].find().to_a
assert output.string.include?("db.test.find") assert output.string.include?("2004")
assert !output.string.include?("db.test.remove")
end end
def test_connection_logger def test_connection_logger