Decodes Avro-encoded `bytes` into a `Row`.
(&mut self, bytes: &mut &[u8])
| 95 | |
| 96 | /// Decodes Avro-encoded `bytes` into a `Row`. |
| 97 | pub async fn decode(&mut self, bytes: &mut &[u8]) -> Result<Result<Row, Error>, Error> { |
| 98 | // Clear out any bytes that might be left over from |
| 99 | // an earlier run. This can happen if the |
| 100 | // `dsr.deserialize` call returns an error, |
| 101 | // causing us to return early. |
| 102 | let mut packer = self.row_buf.packer(); |
| 103 | // The outer Result describes transient errors so use ? here to propagate |
| 104 | let (bytes2, resolved_schema, csr_schema_id) = match self.csr_avro.resolve(bytes).await? { |
| 105 | Ok(ok) => ok, |
| 106 | Err(err) => return Ok(Err(err)), |
| 107 | }; |
| 108 | *bytes = bytes2; |
| 109 | let dec = AvroFlatDecoder { |
| 110 | packer: &mut packer, |
| 111 | buf: &mut self.buf1, |
| 112 | is_top: true, |
| 113 | }; |
| 114 | let dsr = GeneralDeserializer { |
| 115 | schema: resolved_schema.top_node(), |
| 116 | }; |
| 117 | let result = dsr |
| 118 | .deserialize(bytes, dec) |
| 119 | .with_context(|| { |
| 120 | format!( |
| 121 | "unable to decode row {}", |
| 122 | match &csr_schema_id { |
| 123 | Some(id) => format!("(Avro schema id = {:?})", id), |
| 124 | None => "".to_string(), |
| 125 | } |
| 126 | ) |
| 127 | }) |
| 128 | .map(|_| self.row_buf.clone()); |
| 129 | if result.is_ok() { |
| 130 | trace!( |
| 131 | "[customer-data] Decoded row {:?} in {}", |
| 132 | self.row_buf, self.debug_name |
| 133 | ); |
| 134 | } |
| 135 | Ok(result) |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | #[derive(Debug)] |
no test coverage detected