OP_MSG implementation entry point.
(
operation: int,
command: Mapping[str, Any],
docs: list[Mapping[str, Any]],
ack: bool,
opts: CodecOptions[Any],
ctx: _BulkWriteContext,
)
| 874 | |
| 875 | |
| 876 | def _batched_op_msg( |
| 877 | operation: int, |
| 878 | command: Mapping[str, Any], |
| 879 | docs: list[Mapping[str, Any]], |
| 880 | ack: bool, |
| 881 | opts: CodecOptions[Any], |
| 882 | ctx: _BulkWriteContext, |
| 883 | ) -> tuple[int, bytes, list[Mapping[str, Any]]]: |
| 884 | """OP_MSG implementation entry point.""" |
| 885 | buf = _BytesIO() |
| 886 | |
| 887 | # Save space for message length and request id |
| 888 | buf.write(_ZERO_64) |
| 889 | # responseTo, opCode |
| 890 | buf.write(b"\x00\x00\x00\x00\xdd\x07\x00\x00") |
| 891 | |
| 892 | to_send, length = _batched_op_msg_impl(operation, command, docs, ack, opts, ctx, buf) |
| 893 | |
| 894 | # Header - request id and message length |
| 895 | buf.seek(4) |
| 896 | request_id = _randint() |
| 897 | buf.write(_pack_int(request_id)) |
| 898 | buf.seek(0) |
| 899 | buf.write(_pack_int(length)) |
| 900 | |
| 901 | return request_id, buf.getvalue(), to_send |
| 902 | |
| 903 | |
| 904 | if _use_c: |
no test coverage detected