Class: Mongo::Server::ConnectionBase

# File 'lib/mongo/server/connection_base.rb', line 53

attr_reader :options

# File 'lib/mongo/server/connection_base.rb', line 58

attr_reader :server

# File 'lib/mongo/server/connection_base.rb', line 288

def check_timeout!(context)
  return if [nil, 0].include?(context.deadline)

  time_to_execute = context.remaining_timeout_sec - server.minimum_round_trip_time
  if time_to_execute <= 0
    raise Mongo::Error::TimeoutError
  end
end

# File 'lib/mongo/server/connection_base.rb', line 167

def deliver(message, context, options = {})
  if Lint.enabled? && !@socket
    raise Error::LintError, "Trying to deliver a message over a disconnected connection (to #{address})"
  end
  buffer = serialize(message, context)
  check_timeout!(context)
  ensure_connected do |socket|
    operation_id = Monitoring.next_operation_id
    started_event = command_started(address, operation_id, message.payload,
      socket_object_id: socket.object_id, connection_id: id,
      connection_generation: generation,
      server_connection_id: description.server_connection_id,
      service_id: description.service_id,
    )
    start = Utils.monotonic_time
    result = nil
    begin
      result = add_server_diagnostics do
        socket.write(buffer.to_s, timeout: context.remaining_timeout_sec)
        if message.replyable?
          check_timeout!(context)
          Protocol::Message.deserialize(socket, max_message_size, message.request_id, options.merge(timeout: context.remaining_timeout_sec))
        else
          nil
        end
      end
    rescue Exception => e
      total_duration = Utils.monotonic_time - start
      command_failed(nil, address, operation_id, message.payload,
        e.message, total_duration,
        started_event: started_event,
        server_connection_id: description.server_connection_id,
        service_id: description.service_id,
      )
      raise
    else
      total_duration = Utils.monotonic_time - start
      command_completed(result, address, operation_id, message.payload,
        total_duration,
        started_event: started_event,
        server_connection_id: description.server_connection_id,
        service_id: description.service_id,
      )
    end
    if result && context.decrypt?
      result = result.maybe_decrypt(context)
    end
    result
  end
end

# File 'lib/mongo/server/connection_base.rb', line 150

def dispatch(messages, context, options = {})
  # The monitoring code does not correctly handle multiple messages,
  # and the driver internally does not send more than one message at
  # a time ever. Thus prohibit multiple message use for now.
  if messages.length != 1
    raise ArgumentError, 'Can only dispatch one message at a time'
  end
  if description.unknown?
    raise Error::InternalDriverError, "Cannot dispatch a message on a connection with unknown description: #{description.inspect}"
  end
  message = messages.first
  deliver(message, context, options)
end

# File 'lib/mongo/server/connection_base.rb', line 100

def generation
  # If the connection is to a load balancer, @generation is set
  # after handshake completes. If the connection is to another server
  # type, generation is specified during connection creation.
  @generation || options[:generation]
end

# File 'lib/mongo/server/connection_base.rb', line 218

def serialize(message, context, buffer = BSON::ByteBuffer.new)
  # Driver specifications only mandate the fixed 16MiB limit for
  # serialized BSON documents. However, the server returns its
  # active serialized BSON document size limit in the hello response,
  # which is +max_bson_object_size+ below. The +DEFAULT_MAX_BSON_OBJECT_SIZE+
  # is the 16MiB value mandated by the specifications which we use
  # only as the default if the server's hello did not contain
  # maxBsonObjectSize.
  max_bson_size = max_bson_object_size || DEFAULT_MAX_BSON_OBJECT_SIZE
  if context.encrypt?
    # The client-side encryption specification requires bulk writes to
    # be split at a reduced maxBsonObjectSize. If this message is a bulk
    # write and its size exceeds the reduced size limit, the serializer
    # will raise an exception, which is caught by BulkWrite. BulkWrite
    # will split the operation into individual writes, which will
    # not be subject to the reduced maxBsonObjectSize.
    if message.bulk_write?
      # Make the new maximum size equal to the specified reduced size
      # limit plus the 16KiB overhead allowance.
      max_bson_size = REDUCED_MAX_BSON_SIZE
    end
  end

  # RUBY-2234: It is necessary to check that the message size does not
  # exceed the maximum bson object size before compressing and serializing
  # the final message.
  #
  # This is to avoid the case where the user performs a bulk write
  # larger than 16MiB which, when compressed, becomes smaller than 16MiB.
  # If the driver does not split the bulk writes prior to compression,
  # the entire operation will be sent to the server, which will raise an
  # error because the uncompressed operation exceeds the maximum bson size.
  #
  # To address this problem, we serialize the message prior to compression
  # and raise an exception if the serialized message exceeds the maximum
  # bson size.
  if max_message_size
    # Create a separate buffer that contains the un-compressed message
    # for the purpose of checking its size. Write any pre-existing contents
    # from the original buffer into the temporary one.
    temp_buffer = BSON::ByteBuffer.new

    # TODO: address the fact that this line mutates the buffer.
    temp_buffer.put_bytes(buffer.get_bytes(buffer.length))

    message.serialize(temp_buffer, max_bson_size, MAX_BSON_COMMAND_OVERHEAD)
    if temp_buffer.length > max_message_size
      raise Error::MaxMessageSize.new(max_message_size)
    end
  end

  # RUBY-2335: When the un-compressed message is smaller than the maximum
  # bson size limit, the message will be serialized twice. The operations
  # layer should be refactored to allow compression on an already-
  # serialized message.
  final_message = message.maybe_compress(compressor, options[:zlib_compression_level])
  final_message.serialize(buffer, max_bson_size, MAX_BSON_COMMAND_OVERHEAD)

  buffer
end

# File 'lib/mongo/server/connection_base.rb', line 92

def service_id
  description&.service_id
end