1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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
|
# Copyright (C) 2017 MongoDB, 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
module Protocol
# MongoDB Wire protocol Msg message (OP_MSG), a bi-directional wire protocol opcode.
#
# OP_MSG is only available in MongoDB 3.6 (maxWireVersion >= 6) and later.
#
# @api private
#
# @since 2.5.0
class Msg < Message
include Monitoring::Event::Secure
# The identifier for the database name to execute the command on.
#
# @since 2.5.0
DATABASE_IDENTIFIER = '$db'.freeze
# Creates a new OP_MSG protocol message
#
# @example Create a OP_MSG wire protocol message
# Msg.new([:more_to_come], {}, { ismaster: 1 },
# { type: 1, payload: { identifier: 'documents', sequence: [..] } })
#
# @param [ Array<Symbol> ] flags The flag bits. Current supported values are
# :more_to_come and :checksum_present.
# @param [ Hash ] options The options. There are currently no supported options, this is a
# place-holder for the future.
# @param [ BSON::Document, Hash ] global_args The global arguments, becomes a section of payload type 0.
# @param [ BSON::Document, Hash ] sections Zero or more sections, in the format
# { type: 1, payload: { identifier: <String>, sequence: <Array<BSON::Document, Hash>> } } or
# { type: 0, payload: <BSON::Document, Hash> }
#
# @option options [ true, false ] validating_keys Whether keys should be validated.
#
# @api private
#
# @since 2.5.0
def initialize(flags, options, global_args, *sections)
@flags = flags || [ :none ]
@options = options
@global_args = global_args
@sections = [ { type: 0, payload: global_args } ] + sections
@request_id = nil
super
end
# Whether the message expects a reply from the database.
#
# @example Does the message require a reply?
# message.replyable?
#
# @return [ true, false ] If the message expects a reply.
#
# @since 2.5.0
def replyable?
@replyable ||= !flags.include?(:more_to_come)
end
# Return the event payload for monitoring.
#
# @example Return the event payload.
# message.payload
#
# @return [ BSON::Document ] The event payload.
#
# @since 2.5.0
def payload
BSON::Document.new(
command_name: command.keys.first,
database_name: global_args[DATABASE_IDENTIFIER],
command: command,
request_id: request_id,
reply: sections[0]
)
end
# Serializes message into bytes that can be sent on the wire.
#
# @param [ BSON::ByteBuffer ] buffer where the message should be inserted.
# @param [ Integer ] max_bson_size The maximum bson object size.
#
# @return [ BSON::ByteBuffer ] buffer containing the serialized message.
#
# @since 2.5.0
def serialize(buffer = BSON::ByteBuffer.new, max_bson_size = nil)
super
add_check_sum(buffer)
buffer
end
# Compress this message.
#
# @param [ String, Symbol ] compressor The compressor to use.
# @param [ Integer ] zlib_compression_level The zlib compression level to use.
#
# @return [ Compressed, self ] A Protocol::Compressed message or self, depending on whether
# this message can be compressed.
#
# @since 2.5.0
def compress!(compressor, zlib_compression_level = nil)
if compressor && compression_allowed?(command.keys.first)
Compressed.new(self, compressor, zlib_compression_level)
else
self
end
end
private
def command
@command ||= global_args.dup.tap do |cmd|
cmd.delete(DATABASE_IDENTIFIER)
sections.each do |section|
if section[:type] == 1
identifier = section[:payload][:identifier]
cmd[identifier] ||= []
cmd[identifier] += section[:payload][:sequence]
end
end
end
end
def add_check_sum(buffer)
if flags.include?(:checksum_present)
#buffer.put_int32(checksum)
end
end
def global_args
@global_args ||= (sections[0] || {})
end
# The operation code required to specify a OP_MSG message.
# @return [ Fixnum ] the operation code.
#
# @since 2.5.0
OP_CODE = 2013
# Available flags for a OP_MSG message.
FLAGS = Array.new(16).tap { |arr|
arr[0] = :checksum_present
arr[1] = :more_to_come
}
# @!attribute
# @return [Array<Symbol>] The flags for this message.
field :flags, BitVector.new(FLAGS)
# @!attribute
# @return [Hash] The sections of payload type 1 or 0.
field :sections, Sections
alias :documents :sections
Registry.register(OP_CODE, self)
end
end
end
|
