wtx 0.28.0

A collection of different transport implementations and related tools focused primarily on web technologies.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

use crate::{
  database::{
    DatabaseError, StmtCmd,
    client::{
      mysql::{
        ExecutorBuffer, MysqlExecutor, MysqlStatement, MysqlStatementMut, MysqlStatements, Ty,
        column::Column,
        misc::{fetch_protocol, send_packet},
        mysql_protocol::{
          column_res::ColumnRes, prepare_req::PrepareReq, prepare_res::PrepareRes,
          stmt_close_req::StmtCloseReq,
        },
        ty_params::TyParams,
      },
      rdbms::statements_misc::StatementsMisc,
    },
  },
  misc::{
    ArrayString, LeaseMut, Stream, U64String, Vector,
    partitioned_filled_buffer::PartitionedFilledBuffer, u64_string,
  },
};

impl<E, EB, S> MysqlExecutor<E, EB, S>
where
  E: From<crate::Error>,
  EB: LeaseMut<ExecutorBuffer>,
  S: Stream,
{
  #[inline]
  pub(crate) async fn write_send_await_stmt_prot<'stmts, SC>(
    (capabilities, sequence_id): (&mut u64, &mut u8),
    encode_buffer: &mut Vector<u8>,
    net_buffer: &mut PartitionedFilledBuffer,
    sc: SC,
    stmts: &'stmts mut MysqlStatements,
    stream: &mut S,
    cb: impl FnOnce(&mut MysqlStatementMut<'_>) -> Result<(), E>,
  ) -> Result<(u64, U64String, MysqlStatement<'stmts>), E>
  where
    SC: StmtCmd,
  {
    let stmt_cmd_id = sc.hash(stmts.hasher_mut());
    let stmt_cmd_id_array = u64_string(stmt_cmd_id);
    if stmts.get_by_stmt_cmd_id(stmt_cmd_id).is_some() {
      // FIXME(stable): Use `if let Some ...` with polonius
      let mut stmt = stmts.get_by_stmt_cmd_id(stmt_cmd_id).unwrap();
      cb(&mut stmt)?;
      return Ok((stmt_cmd_id, stmt_cmd_id_array, stmt.into()));
    }

    let stmt_cmd = sc.cmd().ok_or_else(|| E::from(DatabaseError::UnknownStatementId.into()))?;

    send_packet(
      (capabilities, sequence_id),
      encode_buffer,
      PrepareReq { query: stmt_cmd.as_bytes() },
      stream,
    )
    .await?;
    let mut builder = stmts
      .builder(((&mut *capabilities, &mut *sequence_id), encode_buffer, &mut *stream), {
        async fn fun<S>(
          ((capabilities, sequence_id), encode_buffer, stream): &mut (
            (&mut u64, &mut u8),
            &mut Vector<u8>,
            &mut S,
          ),
          stmt: StatementsMisc<u32>,
        ) -> crate::Result<()>
        where
          S: Stream,
        {
          send_packet(
            (capabilities, sequence_id),
            encode_buffer,
            StmtCloseReq { statement: stmt._aux },
            stream,
          )
          .await
        }
        fun
      })
      .await?;
    let pres: PrepareRes = fetch_protocol(*capabilities, net_buffer, sequence_id, stream).await?.0;
    let _ = builder.expand(pres.params.max(pres.columns).into(), dummy())?;
    let elements = builder.inserted_elements();
    for _ in 0..pres.params {
      let _: ColumnRes = fetch_protocol(*capabilities, net_buffer, sequence_id, stream).await?.0;
    }
    for idx in 0..pres.columns {
      let cres = fetch_protocol(*capabilities, net_buffer, sequence_id, stream).await?.0;
      let Some(elem) = elements.get_mut(usize::from(idx)) else {
        break;
      };
      elem.0 = Column::from_column_res(&cres);
    }
    let sm = StatementsMisc::new(pres.statement_id, pres.columns.into(), 0);
    let idx = builder.build(stmt_cmd_id, sm)?;
    let Some(mut stmt) = stmts.get_by_idx(idx) else {
      return Err(crate::Error::ProgrammingError.into());
    };
    cb(&mut stmt)?;
    Ok((stmt_cmd_id, stmt_cmd_id_array, stmt.into()))
  }
}

#[inline]
fn dummy() -> (Column, TyParams) {
  (
    Column { name: ArrayString::new(), ty_params: TyParams { flags: 0, ty: Ty::Bit } },
    TyParams { flags: 0, ty: Ty::Bit },
  )
}