Internal compressed OP_MSG message helper.
(
flags: int,
command: Mapping[str, Any],
identifier: str,
docs: Optional[list[Mapping[str, Any]]],
opts: CodecOptions[Any],
)
| 375 | |
| 376 | |
| 377 | def _op_msg_uncompressed( |
| 378 | flags: int, |
| 379 | command: Mapping[str, Any], |
| 380 | identifier: str, |
| 381 | docs: Optional[list[Mapping[str, Any]]], |
| 382 | opts: CodecOptions[Any], |
| 383 | ) -> tuple[int, bytes, int, int]: |
| 384 | """Internal compressed OP_MSG message helper.""" |
| 385 | data, total_size, max_bson_size = _op_msg_no_header(flags, command, identifier, docs, opts) |
| 386 | request_id, op_message = __pack_message(2013, data) |
| 387 | return request_id, op_message, total_size, max_bson_size |
| 388 | |
| 389 | |
| 390 | if _use_c: |
no test coverage detected