OP_MSG implementation entry point for client-level bulkWrite.
(
command: Mapping[str, Any],
operations: list[tuple[str, Mapping[str, Any]]],
namespaces: list[str],
ack: bool,
opts: CodecOptions[Any],
ctx: _ClientBulkWriteContext,
)
| 1196 | |
| 1197 | |
| 1198 | def _client_batched_op_msg( |
| 1199 | command: Mapping[str, Any], |
| 1200 | operations: list[tuple[str, Mapping[str, Any]]], |
| 1201 | namespaces: list[str], |
| 1202 | ack: bool, |
| 1203 | opts: CodecOptions[Any], |
| 1204 | ctx: _ClientBulkWriteContext, |
| 1205 | ) -> tuple[int, bytes, list[Mapping[str, Any]], list[Mapping[str, Any]]]: |
| 1206 | """OP_MSG implementation entry point for client-level bulkWrite.""" |
| 1207 | buf = _BytesIO() |
| 1208 | |
| 1209 | # Save space for message length and request id |
| 1210 | buf.write(_ZERO_64) |
| 1211 | # responseTo, opCode |
| 1212 | buf.write(b"\x00\x00\x00\x00\xdd\x07\x00\x00") |
| 1213 | |
| 1214 | to_send_ops, to_send_ns, length = _client_batched_op_msg_impl( |
| 1215 | command, operations, namespaces, ack, opts, ctx, buf |
| 1216 | ) |
| 1217 | |
| 1218 | # Header - request id and message length |
| 1219 | buf.seek(4) |
| 1220 | request_id = _randint() |
| 1221 | buf.write(_pack_int(request_id)) |
| 1222 | buf.seek(0) |
| 1223 | buf.write(_pack_int(length)) |
| 1224 | |
| 1225 | return request_id, buf.getvalue(), to_send_ops, to_send_ns |
| 1226 | |
| 1227 | |
| 1228 | def _client_do_batched_op_msg( |
no test coverage detected