Persist a subscriber cursor.
(&self, state: &ArraySubscriberState)
| 202 | |
| 203 | /// Persist a subscriber cursor. |
| 204 | fn save(&self, state: &ArraySubscriberState) -> crate::Result<()> { |
| 205 | let key = Self::cursor_key(&state.session_id, &state.array_name); |
| 206 | let bytes = zerompk::to_msgpack_vec(state).map_err(|e| crate::Error::Storage { |
| 207 | engine: "array_sync".into(), |
| 208 | detail: format!("subscriber_store save encode: {e}"), |
| 209 | })?; |
| 210 | let txn = self.db.begin_write().map_err(|e| crate::Error::Storage { |
| 211 | engine: "array_sync".into(), |
| 212 | detail: format!("subscriber_store save begin_write: {e}"), |
| 213 | })?; |
| 214 | { |
| 215 | let mut table = txn |
| 216 | .open_table(CURSOR_TABLE) |
| 217 | .map_err(|e| crate::Error::Storage { |
| 218 | engine: "array_sync".into(), |
| 219 | detail: format!("subscriber_store save open_table: {e}"), |
| 220 | })?; |
| 221 | table |
| 222 | .insert(key.as_str(), bytes.as_slice()) |
| 223 | .map_err(|e| crate::Error::Storage { |
| 224 | engine: "array_sync".into(), |
| 225 | detail: format!("subscriber_store save insert: {e}"), |
| 226 | })?; |
| 227 | } |
| 228 | txn.commit().map_err(|e| crate::Error::Storage { |
| 229 | engine: "array_sync".into(), |
| 230 | detail: format!("subscriber_store save commit: {e}"), |
| 231 | })?; |
| 232 | Ok(()) |
| 233 | } |
| 234 | |
| 235 | /// Load a subscriber cursor, returning `None` if not found. |
| 236 | fn load(&self, session_id: &str, array_name: &str) -> Option<ArraySubscriberState> { |