Internal OP_MSG message helper.
(
flags: int,
command: Mapping[str, Any],
identifier: str,
docs: Optional[list[Mapping[str, Any]]],
opts: CodecOptions[Any],
ctx: Union[SnappyContext, ZlibContext, ZstdContext],
)
| 361 | |
| 362 | |
| 363 | def _op_msg_compressed( |
| 364 | flags: int, |
| 365 | command: Mapping[str, Any], |
| 366 | identifier: str, |
| 367 | docs: Optional[list[Mapping[str, Any]]], |
| 368 | opts: CodecOptions[Any], |
| 369 | ctx: Union[SnappyContext, ZlibContext, ZstdContext], |
| 370 | ) -> tuple[int, bytes, int, int]: |
| 371 | """Internal OP_MSG message helper.""" |
| 372 | msg, total_size, max_bson_size = _op_msg_no_header(flags, command, identifier, docs, opts) |
| 373 | rid, msg = _compress(2013, msg, ctx) |
| 374 | return rid, msg, total_size, max_bson_size |
| 375 | |
| 376 | |
| 377 | def _op_msg_uncompressed( |
no test coverage detected