Method send_protocol_pump

src/pglite/postgres_mod.rs:1192–1229

(
        &mut self,
        payload: &[u8],
        continuation_prefix: impl FnOnce() -> Vec<u8>,
    )

    }

    pub(crate) fn send_protocol_pump(
        &mut self,
        payload: &[u8],
        continuation_prefix: impl FnOnce() -> Vec<u8>,
    ) -> Result<ProtocolPumpOutcome> {
        {
            let _phase = timing::phase("postgres.protocol.ensure_started");
            self.start_protocol()?;
        }
        if payload.is_empty() {
            return Ok(ProtocolPumpOutcome::Buffered(Vec::new()));
        }
        ensure!(
            self.protocol_stdio_attachment.is_some(),
            "WASIX protocol pump requires an attached stream"
        );
        let previous_mode = self.set_protocol_transport(ProtocolTransportMode::Hybrid)?;
        ensure!(
            previous_mode == ProtocolTransportMode::Buffered,
            "WASIX protocol transport was not buffered before protocol pump"
        );
        let result = self.send_protocol_inner(payload);
        let active = self.protocol_stream_active().unwrap_or(false);
        if active {
            self.set_protocol_stream_prefix(continuation_prefix())?;
            let stream_result = result.and_then(|_| self.serve_protocol_stream_inner());
            let restore_result = self.restore_protocol_transport(previous_mode);
            let clear_result = self.clear_protocol_stream_prefix();
            stream_result.and(restore_result).and(clear_result)?;
            Ok(ProtocolPumpOutcome::Streamed)
        } else {
            let output = result;
            let restore_result = self.restore_protocol_transport(previous_mode);
            restore_result?;
            let output = output?;
            Ok(ProtocolPumpOutcome::Buffered(output))
        }
    }

    fn send_protocol_inner(&mut self, payload: &[u8]) -> Result<Vec<u8>> {
        self.reset_backend_c_timings()?;