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

Finished removing Message classes. Made Cursor#close threadsafe.

This commit is contained in:
Kyle Banker 2009-10-21 10:11:33 -04:00
parent b938765ac0
commit de5c078cec
15 changed files with 64 additions and 317 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Rakefile
View File

@ -18,7 +18,7 @@ gem_command = "gem1.9" if $0.match(/1\.9$/) # use gem1.9 if we used rake1.9
desc "Test the MongoDB Ruby driver." desc "Test the MongoDB Ruby driver."
Rake::TestTask.new(:test) do |t| Rake::TestTask.new(:test) do |t|
t.test_files = FileList['test/test*.rb'] t.test_files = FileList['test/test*.rb']
t.verbose = true t.verbose = true
end end
desc "Generate documentation" desc "Generate documentation"

2
lib/mongo.rb
View File

@ -6,8 +6,8 @@ require 'mongo/types/regexp_of_holding'
require 'mongo/util/conversions' require 'mongo/util/conversions'
require 'mongo/errors' require 'mongo/errors'
require 'mongo/constants'
require 'mongo/connection' require 'mongo/connection'
require 'mongo/message'
require 'mongo/db' require 'mongo/db'
require 'mongo/cursor' require 'mongo/cursor'
require 'mongo/collection' require 'mongo/collection'

8
lib/mongo/collection.rb
View File

@ -229,7 +229,7 @@ module Mongo
BSON.serialize_cstr(message, "#{@db.name}.#{@name}") BSON.serialize_cstr(message, "#{@db.name}.#{@name}")
message.put_int(0) message.put_int(0)
message.put_array(BSON.new.serialize(selector, false).to_a) message.put_array(BSON.new.serialize(selector, false).to_a)
@db.send_message_with_operation(OP_DELETE, message) @db.send_message_with_operation(Mongo::Constants::OP_DELETE, message)
end end
# Remove all records. # Remove all records.
@ -259,7 +259,7 @@ module Mongo
message.put_int(options[:upsert] ? 1 : 0) # 1 if a repsert operation (upsert) message.put_int(options[:upsert] ? 1 : 0) # 1 if a repsert operation (upsert)
message.put_array(BSON.new.serialize(spec, false).to_a) message.put_array(BSON.new.serialize(spec, false).to_a)
message.put_array(BSON.new.serialize(document, false).to_a) message.put_array(BSON.new.serialize(document, false).to_a)
@db.send_message_with_operation(OP_UPDATE, message) @db.send_message_with_operation(Mongo::Constants::OP_UPDATE, message)
if options[:safe] && error=@db.error if options[:safe] && error=@db.error
raise OperationFailure, error raise OperationFailure, error
@ -454,7 +454,7 @@ EOS
private private
# Sends an OP_INSERT message to the database. # Sends an Mongo::Constants::OP_INSERT message to the database.
# Takes an array of +documents+, an optional +collection_name+, and a # Takes an array of +documents+, an optional +collection_name+, and a
# +check_keys+ setting. # +check_keys+ setting.
def insert_documents(documents, collection_name=@name, check_keys=true) def insert_documents(documents, collection_name=@name, check_keys=true)
@ -462,7 +462,7 @@ EOS
message.put_int(0) message.put_int(0)
BSON.serialize_cstr(message, "#{@db.name}.#{collection_name}") BSON.serialize_cstr(message, "#{@db.name}.#{collection_name}")
documents.each { |doc| message.put_array(BSON.new.serialize(doc, check_keys).to_a) } documents.each { |doc| message.put_array(BSON.new.serialize(doc, check_keys).to_a) }
@db.send_message_with_operation(OP_INSERT, message) @db.send_message_with_operation(Mongo::Constants::OP_INSERT, message)
documents.collect { |o| o[:_id] || o['_id'] } documents.collect { |o| o[:_id] || o['_id'] }
end end

15
lib/mongo/constants.rb Normal file
View File

@ -0,0 +1,15 @@
module Mongo
module Constants
OP_REPLY = 1
OP_MSG = 1000
OP_UPDATE = 2001
OP_INSERT = 2002
OP_QUERY = 2004
OP_GET_MORE = 2005
OP_DELETE = 2006
OP_KILL_CURSORS = 2007
OP_QUERY_SLAVE_OK = 4
OP_QUERY_NO_CURSOR_TIMEOUT = 16
end
end

41
lib/mongo/cursor.rb
View File

@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
require 'mongo/message'
require 'mongo/util/byte_buffer' require 'mongo/util/byte_buffer'
require 'mongo/util/bson' require 'mongo/util/bson'
@ -173,17 +172,23 @@ module Mongo
# Close the cursor. # Close the cursor.
# #
# Note: if a cursor is read until exhausted (read until OP_QUERY or # Note: if a cursor is read until exhausted (read until Mongo::Constants::OP_QUERY or
# OP_GETMORE returns zero for the cursor id), there is no need to # Mongo::Constants::OP_GETMORE returns zero for the cursor id), there is no need to
# close it by calling this method. # close it by calling this method.
# #
# Collection#find takes an optional block argument which can be used to # Collection#find takes an optional block argument which can be used to
# ensure that your cursors get closed. See the documentation for # ensure that your cursors get closed. See the documentation for
# Collection#find for details. # Collection#find for details.
def close def close
@db.send_to_db(KillCursorsMessage.new(@cursor_id)) if @cursor_id if @cursor_id
message = ByteBuffer.new
message.put_int(0)
message.put_int(1)
message.put_long(@cursor_id)
@db.send_message_with_operation(Mongo::Constants::OP_KILL_CURSORS, message)
end
@cursor_id = 0 @cursor_id = 0
@closed = true @closed = true
end end
# Returns true if this cursor is closed, false otherwise. # Returns true if this cursor is closed, false otherwise.
@ -204,7 +209,16 @@ module Mongo
end end
def read_message_header def read_message_header
MessageHeader.new.read_header(@db) message = ByteBuffer.new
message.put_array(@db.receive_full(16).unpack("C*"))
unless message.size == 16 #HEADER_SIZE
raise "Short read for DB response header: expected #{16} bytes, saw #{message.size}"
end
message.rewind
size = message.get_int
request_id = message.get_int
response_to = message.get_int
op = message.get_int
end end
def read_response_header def read_response_header
@ -221,9 +235,6 @@ module Mongo
else else
@n_received = @n_remaining @n_received = @n_remaining
end end
if @query.number_to_return > 0 and @n_received >= @query.number_to_return
close()
end
end end
def num_remaining def num_remaining
@ -262,9 +273,10 @@ module Mongo
# Cursor id. # Cursor id.
message.put_long(@cursor_id) message.put_long(@cursor_id)
@db.send_message_with_operation_without_synchronize(OP_GET_MORE, message) @db.send_message_with_operation_without_synchronize(Mongo::Constants::OP_GET_MORE, message)
read_all read_all
} }
close_cursor_if_query_complete
end end
def object_from_stream def object_from_stream
@ -283,12 +295,13 @@ module Mongo
if @query_run if @query_run
false false
else else
message = construct_query_message(@query)
@db._synchronize { @db._synchronize {
message = construct_query_message(@query) @db.send_message_with_operation_without_synchronize(Mongo::Constants::OP_QUERY, message)
@query_run = true @query_run = true
@db.send_message_with_operation_without_synchronize(OP_QUERY, message)
read_all read_all
} }
close_cursor_if_query_complete
true true
end end
end end
@ -337,6 +350,10 @@ module Mongo
"DBResponse(flags=#@result_flags, cursor_id=#@cursor_id, start=#@starting_from)" "DBResponse(flags=#@result_flags, cursor_id=#@cursor_id, start=#@starting_from)"
end end
def close_cursor_if_query_complete
close if @query.number_to_return > 0 && @n_received >= @query.number_to_return
end
def check_modifiable def check_modifiable
if @query_run || @closed if @query_run || @closed
raise InvalidOperation, "Cannot modify the query once it has been run or closed." raise InvalidOperation, "Cannot modify the query once it has been run or closed."

26
lib/mongo/db.rb
View File

@ -18,7 +18,6 @@ require 'socket'
require 'digest/md5' require 'digest/md5'
require 'thread' require 'thread'
require 'mongo/collection' require 'mongo/collection'
require 'mongo/message'
require 'mongo/query' require 'mongo/query'
require 'mongo/util/ordered_hash.rb' require 'mongo/util/ordered_hash.rb'
require 'mongo/admin' require 'mongo/admin'
@ -358,11 +357,6 @@ module Mongo
message message
end end
# Send a MsgMessage to the database.
def send_message(msg)
send_to_db(MsgMessage.new(msg))
end
# Returns a Cursor over the query results. # Returns a Cursor over the query results.
# #
# Note that the query gets sent lazily; the cursor calls # Note that the query gets sent lazily; the cursor calls
@ -470,16 +464,16 @@ module Mongo
end end
def send_message_with_operation_without_synchronize(operation, message) def send_message_with_operation_without_synchronize(operation, message)
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)
@logger.debug(" MONGODB #{operation} #{message}") if @logger @logger.debug(" MONGODB #{operation} #{message}") if @logger
@socket.print(message_with_headers.to_s) @socket.print(message_with_headers.to_s)
@socket.flush @socket.flush
rescue => ex rescue => ex
close close
raise ex raise ex
end end
end end
def receive_message_with_operation(operation, message) def receive_message_with_operation(operation, message)

20
lib/mongo/message.rb
View File

@ -1,20 +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.
# ++
%w(kill_cursors_message message_header
msg_message).each { |f|
require "mongo/message/#{f}"
}

31
lib/mongo/message/kill_cursors_message.rb
View File

@ -1,31 +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 KillCursorsMessage < Message
def initialize(*cursors)
super(OP_KILL_CURSORS)
write_int(0)
write_int(cursors.length)
cursors.each { |c| write_long c }
end
end
end

80
lib/mongo/message/message.rb
View File

@ -1,80 +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/util/bson'
require 'mongo/util/byte_buffer'
module Mongo
class Message
HEADER_SIZE = 16 # size, id, response_to, opcode
@@class_req_id = 0
attr_reader :buf # for testing
def initialize(op)
@op = op
@message_length = HEADER_SIZE
@data_length = 0
@request_id = (@@class_req_id += 1)
@response_id = 0
@buf = ByteBuffer.new
@buf.put_int(16) # holder for length
@buf.put_int(@request_id)
@buf.put_int(0) # response_to
@buf.put_int(op)
end
def write_int(i)
@buf.put_int(i)
update_message_length
end
def write_long(i)
@buf.put_long(i)
update_message_length
end
def write_string(s)
BSON.serialize_cstr(@buf, s)
update_message_length
end
def write_doc(hash, check_keys=false)
@buf.put_array(BSON.new.serialize(hash, check_keys).to_a)
update_message_length
end
def to_a
@buf.to_a
end
def dump
@buf.dump
end
# Do not call. Private, but kept public for testing.
def update_message_length
pos = @buf.position
@buf.put_int(@buf.size, 0)
@buf.position = pos
end
end
end

45
lib/mongo/message/message_header.rb
View File

@ -1,45 +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/util/byte_buffer'
module Mongo
class MessageHeader
HEADER_SIZE = 16
def initialize()
@buf = ByteBuffer.new
end
def read_header(db)
@buf.rewind
@buf.put_array(db.receive_full(HEADER_SIZE).unpack("C*"))
raise "Short read for DB response header: expected #{HEADER_SIZE} bytes, saw #{@buf.size}" unless @buf.size == HEADER_SIZE
@buf.rewind
@size = @buf.get_int
@request_id = @buf.get_int
@response_to = @buf.get_int
@op = @buf.get_int
self
end
def dump
@buf.dump
end
end
end

29
lib/mongo/message/msg_message.rb
View File

@ -1,29 +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 MsgMessage < Message
def initialize(msg)
super(OP_MSG)
write_string(msg)
end
end
end

30
lib/mongo/message/opcodes.rb
View File

@ -1,30 +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.
# ++
module Mongo
OP_REPLY = 1 # reply. responseTo is set.
OP_MSG = 1000 # generic msg command followed by a string
OP_UPDATE = 2001 # update object
OP_INSERT = 2002
# GET_BY_OID = 2003
OP_QUERY = 2004
OP_GET_MORE = 2005
OP_DELETE = 2006
OP_KILL_CURSORS = 2007
OP_QUERY_SLAVE_OK = 4
OP_QUERY_NO_CURSOR_TIMEOUT = 16
end

7
lib/mongo/query.rb
View File

@ -15,7 +15,6 @@
# ++ # ++
require 'mongo/collection' require 'mongo/collection'
require 'mongo/message'
require 'mongo/types/code' require 'mongo/types/code'
module Mongo module Mongo
@ -119,10 +118,10 @@ module Mongo
end end
# Returns an integer indicating which query options have been selected. # Returns an integer indicating which query options have been selected.
# See http://www.mongodb.org/display/DOCS/Mongo+Wire+Protocol#MongoWireProtocol-OPQUERY # See http://www.mongodb.org/display/DOCS/Mongo+Wire+Protocol#MongoWireProtocol-Mongo::Constants::OPQUERY
def query_opts def query_opts
timeout = @timeout ? 0 : OP_QUERY_NO_CURSOR_TIMEOUT timeout = @timeout ? 0 : Mongo::Constants::OP_QUERY_NO_CURSOR_TIMEOUT
slave_ok = @slave_ok ? OP_QUERY_SLAVE_OK : 0 slave_ok = @slave_ok ? Mongo::Constants::OP_QUERY_SLAVE_OK : 0
slave_ok + timeout slave_ok + timeout
end end

10
mongo-ruby-driver.gemspec
View File

@ -20,20 +20,13 @@ PACKAGE_FILES = ['README.rdoc', 'Rakefile', 'mongo-ruby-driver.gemspec',
'lib/mongo/admin.rb', 'lib/mongo/admin.rb',
'lib/mongo/collection.rb', 'lib/mongo/collection.rb',
'lib/mongo/connection.rb', 'lib/mongo/connection.rb',
'lib/mongo/constants.rb',
'lib/mongo/cursor.rb', 'lib/mongo/cursor.rb',
'lib/mongo/db.rb', 'lib/mongo/db.rb',
'lib/mongo/gridfs/chunk.rb', 'lib/mongo/gridfs/chunk.rb',
'lib/mongo/gridfs/grid_store.rb', 'lib/mongo/gridfs/grid_store.rb',
'lib/mongo/gridfs.rb', 'lib/mongo/gridfs.rb',
'lib/mongo/errors.rb', 'lib/mongo/errors.rb',
'lib/mongo/message/get_more_message.rb',
'lib/mongo/message/kill_cursors_message.rb',
'lib/mongo/message/message.rb',
'lib/mongo/message/message_header.rb',
'lib/mongo/message/msg_message.rb',
'lib/mongo/message/opcodes.rb',
'lib/mongo/message/query_message.rb',
'lib/mongo/message.rb',
'lib/mongo/query.rb', 'lib/mongo/query.rb',
'lib/mongo/types/binary.rb', 'lib/mongo/types/binary.rb',
'lib/mongo/types/code.rb', 'lib/mongo/types/code.rb',
@ -72,7 +65,6 @@ TEST_FILES = ['test/mongo-qa/_common.rb',
'test/test_db_api.rb', 'test/test_db_api.rb',
'test/test_db_connection.rb', 'test/test_db_connection.rb',
'test/test_grid_store.rb', 'test/test_grid_store.rb',
'test/test_message.rb',
'test/test_objectid.rb', 'test/test_objectid.rb',
'test/test_ordered_hash.rb', 'test/test_ordered_hash.rb',
'test/test_threading.rb', 'test/test_threading.rb',

35
test/test_message.rb
View File

@ -1,35 +0,0 @@
$LOAD_PATH[0,0] = File.join(File.dirname(__FILE__), '..', 'lib')
require 'mongo'
require 'test/unit'
class MessageTest < Test::Unit::TestCase
include Mongo
def setup
@msg = Message.new(42)
end
def test_initial_info
assert_equal Message::HEADER_SIZE, @msg.buf.length
@msg.write_long(1029)
@msg.buf.rewind
assert_equal Message::HEADER_SIZE + 8, @msg.buf.get_int
@msg.buf.get_int # skip message id
assert_equal 0, @msg.buf.get_int
assert_equal 42, @msg.buf.get_int
assert_equal 1029, @msg.buf.get_long
end
def test_update_length
@msg.update_message_length
@msg.buf.rewind
assert_equal Message::HEADER_SIZE, @msg.buf.get_int
end
def test_long_length
@msg.write_long(1027)
assert_equal Message::HEADER_SIZE + 8, @msg.buf.length
end
end