1#![deny(unsafe_code)]
2
3use std::convert::From as _;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use bytes::Bytes;
9use futures::channel::{mpsc, oneshot};
10use futures::{SinkExt, StreamExt};
11use serde_json::{self, json};
12
13use rmqtt::{
14 fitter::Fitter,
15 hook::{Handler, HookResult, Parameter, Register, ReturnType, Type},
16 plugin::{PackageInfo, Plugin},
17 register,
18 session::Session,
19 types::DisconnectInfo,
20 types::{ClientId, From, Publish, SessionSubMap, SessionSubs, TimestampMillis},
21 utils::timestamp_millis,
22 Result,
23};
24
25use rmqtt_storage::{init_db, DefaultStorageDB, List, Map, StorageType};
26
27use config::PluginConfig;
28use rmqtt::context::ServerContext;
29use rmqtt::inflight::OutInflightMessage;
30use rmqtt::macros::Plugin;
31use rmqtt::session::SessionState;
32use session::{Basic, StorageSessionManager, StoredSessionInfo, StoredSessionInfos};
33use session::{StoredKey, BASIC, DISCONNECT_INFO, INFLIGHT_MESSAGES, LAST_TIME, SESSION_SUB_MAP};
34
35mod config;
36mod session;
37
38enum RebuildChanType {
39 Session(Session, Duration),
40 Done(oneshot::Sender<()>),
41}
42
43type OfflineMessageOptionType = Option<(ClientId, From, Publish)>;
44
45register!(StoragePlugin::new);
46
47#[derive(Plugin)]
48struct StoragePlugin {
49 scx: ServerContext,
50 cfg: Arc<PluginConfig>,
51 storage_db: DefaultStorageDB,
52 stored_session_infos: StoredSessionInfos,
53 register: Box<dyn Register>,
54 session_mgr: StorageSessionManager,
55 rebuild_tx: mpsc::Sender<RebuildChanType>,
56}
57
58impl StoragePlugin {
59 #[inline]
60 async fn new<S: Into<String>>(scx: ServerContext, name: S) -> Result<Self> {
61 let name = name.into();
62 let mut cfg = scx.plugins.read_config::<PluginConfig>(&name)?;
63 match cfg.storage.typ {
64 #[cfg(feature = "sled")]
65 StorageType::Sled => {
66 cfg.storage.sled.path =
67 cfg.storage.sled.path.replace("{node}", &format!("{}", scx.node.id()));
68 }
69 #[cfg(feature = "redis")]
70 StorageType::Redis => {
71 cfg.storage.redis.prefix =
72 cfg.storage.redis.prefix.replace("{node}", &format!("{}", scx.node.id()));
73 }
74 #[cfg(feature = "redis-cluster")]
75 StorageType::RedisCluster => {
76 cfg.storage.redis_cluster.prefix =
77 cfg.storage.redis_cluster.prefix.replace("{node}", &format!("{}", scx.node.id()));
78 }
79 }
80
81 log::info!("{name} StoragePlugin cfg: {cfg:?}");
82
83 let storage_db = match init_db(&cfg.storage).await {
84 Err(e) => {
85 log::error!("{name} init storage db error, {e}");
86 return Err(e);
87 }
88 Ok(db) => db,
89 };
90
91 let stored_session_infos = StoredSessionInfos::new();
92
93 let register = scx.extends.hook_mgr().register();
94 let session_mgr = StorageSessionManager::new(storage_db.clone(), stored_session_infos.clone());
95
96 let cfg = Arc::new(cfg);
97 let rebuild_tx = Self::start_local_runtime(scx.clone());
98 Ok(Self { scx, cfg, storage_db, stored_session_infos, register, session_mgr, rebuild_tx })
99 }
100
101 async fn load_offline_session_infos(&mut self) -> Result<()> {
102 log::info!("{:?} load_offline_session_infos ...", self.name());
103 let storage_db = self.storage_db.clone();
104 let mut iter_storage_db = storage_db.clone();
105 let mut map_iter = iter_storage_db.map_iter().await?;
107 while let Some(m) = map_iter.next().await {
108 match m {
109 Ok(m) => {
110 let id_key = StoredKey::from(map_stored_key_to_id_bytes(m.name()).to_vec());
111 log::debug!("map_stored_key: {id_key:?}");
112 let basic = match m.get::<_, Basic>(BASIC).await {
113 Err(e) => {
114 log::warn!("{id_key:?} load offline session basic info error, {e:?}");
115 if let Err(e) = storage_db.map_remove(m.name()).await {
116 log::warn!("{id_key:?} remove offline session info error, {e:?}");
117 }
118 continue;
119 }
120 Ok(None) => {
121 log::warn!("{id_key:?} offline session basic info is None");
122 if let Err(e) = storage_db.map_remove(m.name()).await {
123 log::warn!("{id_key:?} remove offline session info error, {e:?}");
124 }
125 continue;
126 }
127 Ok(Some(basic)) => basic,
128 };
129
130 log::debug!("basic: {basic:?}");
131 log::debug!("map key: {id_key:?}");
132 let mut s_info = StoredSessionInfo::from(id_key.clone(), basic);
133
134 match m.get::<_, TimestampMillis>(LAST_TIME).await {
135 Ok(Some(last_time)) => {
136 log::debug!("last_time: {last_time:?}");
137 s_info.set_last_time(last_time);
138 }
139 Ok(None) => {}
140 Err(e) => {
141 log::warn!("{id_key:?} load offline session last time error, {e:?}");
142 }
143 }
144
145 match m.get::<_, SessionSubMap>(SESSION_SUB_MAP).await {
146 Ok(Some(subs)) => {
147 log::debug!("subs: {subs:?}");
148 s_info.set_subs(subs);
149 }
150 Ok(None) => {}
151 Err(e) => {
152 log::warn!("{id_key:?} load offline session subscription info error, {e:?}");
153 }
154 }
155
156 match m.get::<_, DisconnectInfo>(DISCONNECT_INFO).await {
157 Ok(Some(disc_info)) => {
158 log::debug!("disc_info: {disc_info:?}");
159 s_info.set_disconnect_info(disc_info);
160 }
161 Ok(None) => {}
162 Err(e) => {
163 log::warn!("{id_key:?} load offline session disconnect info error, {e:?}");
164 }
165 }
166
167 match m.get::<_, Vec<OutInflightMessage>>(INFLIGHT_MESSAGES).await {
168 Ok(Some(inflights)) => {
169 log::debug!("inflights len: {:?}", inflights.len());
170 s_info.inflight_messages = inflights;
171 }
172 Ok(None) => {}
173 Err(e) => {
174 log::warn!("{id_key:?} load offline session inflight messages error, {e:?}");
175 }
176 }
177
178 self.stored_session_infos.add(s_info);
179 }
180 Err(e) => {
181 log::warn!("load offline session info error, {e:?}");
182 }
183 }
184 }
185 drop(map_iter);
186
187 let mut list_iter = iter_storage_db.list_iter().await?;
188 while let Some(l) = list_iter.next().await {
189 match l {
190 Ok(l) => {
191 let id_key = StoredKey::from(list_stored_key_to_id_bytes(l.name()).to_vec());
192 log::debug!("list_stored_key, id_key: {id_key:?}");
193 match l.all::<OfflineMessageOptionType>().await {
194 Ok(offline_msgs) => {
195 log::debug!("{:?} offline_msgs len: {}", id_key, offline_msgs.len(),);
196 let ok =
197 self.stored_session_infos.set_offline_messages(id_key.clone(), offline_msgs);
198 log::debug!("{id_key:?} stored_session_infos, set_offline_messages res: {ok}");
199 if !ok {
200 if let Err(e) = storage_db.list_remove(l.name()).await {
201 log::warn!("{id_key:?} remove offline messages error, {e:?}");
202 }
203 }
204 }
205 Err(e) => {
206 log::warn!("{id_key:?} load offline messages error, {e:?}");
207 if let Err(e) = storage_db.list_remove(l.name()).await {
208 log::warn!("{id_key:?} remove offline messages error, {e:?}");
209 }
210 }
211 }
212 }
213 Err(e) => {
214 log::warn!("load offline messages error, {e:?}");
215 }
216 }
217 }
218 drop(list_iter);
219
220 for removed_key in self.stored_session_infos.retain_latests() {
221 storage_db.map_remove(make_map_stored_key(removed_key.as_ref())).await?;
222 storage_db.list_remove(make_list_stored_key(removed_key.as_ref())).await?;
223 }
224 log::info!("stored_session_infos len: {:?}", self.stored_session_infos.len());
225
226 Ok(())
227 }
228
229 fn start_local_runtime(scx: ServerContext) -> mpsc::Sender<RebuildChanType> {
230 let (tx, mut rx) = futures::channel::mpsc::channel::<RebuildChanType>(100_000);
231 std::thread::spawn(move || {
232 let local_rt = tokio::runtime::Builder::new_current_thread()
233 .enable_all()
234 .build()
235 .expect("tokio runtime build failed");
236 let local_set = tokio::task::LocalSet::new();
237
238 local_set.block_on(&local_rt, async {
239 let exec = scx.get_exec("SESSION_REBUILD_EXEC");
240 while let Some(msg) = rx.next().await {
241 match msg {
242 RebuildChanType::Session(session, session_expiry_interval) => {
243 match SessionState::offline_restart(session.clone(), session_expiry_interval).await {
244 Err(e) => {
245 log::warn!("Rebuild offline sessions error, {e:?}");
246 },
247 Ok(msg_tx) => {
248 let mut session_entry =
249 scx.extends.shared().await.entry(session.id.clone());
250
251 let id = session_entry.id().clone();
252 let task_fut = async move {
253 if let Err(e) = session_entry.set(session, msg_tx).await {
254 log::warn!("{:?} Rebuild offline sessions error, {:?}", session_entry.id(), e);
255 }
256 };
257 if let Err(e) = exec.spawn(task_fut).await {
258 log::warn!("{:?} Rebuild offline sessions error, {:?}", id, e.to_string());
259 }
260
261 let completed_count = exec.completed_count().await;
262 if completed_count > 0 && completed_count % 5000 == 0 {
263 log::info!(
264 "{:?} Rebuild offline sessions, completed_count: {}, active_count: {}, waiting_count: {}, rate: {:?}",
265 id,
266 exec.completed_count().await, exec.active_count(), exec.waiting_count(), exec.rate().await
267 );
268 }
269 }
270 }
271 },
272 RebuildChanType::Done(done_tx) => {
273 let _ = exec.flush().await;
274 let _ = done_tx.send(());
275 log::info!(
276 "Rebuild offline sessions, completed_count: {}, active_count: {}, waiting_count: {}, rate: {:?}",
277 exec.completed_count().await, exec.active_count(), exec.waiting_count(), exec.rate().await
278 );
279 }
280 }
281 }
282 });
283 log::info!("Offline session rebuilding finished");
284 });
285 tx
286 }
287}
288
289#[async_trait]
290impl Plugin for StoragePlugin {
291 #[inline]
292 async fn init(&mut self) -> Result<()> {
293 log::info!("{} init", self.name());
294 self.register
295 .add(
296 Type::BeforeStartup,
297 Box::new(StorageHandler::new(
298 self.scx.clone(),
299 self.storage_db.clone(),
300 self.cfg.clone(),
301 self.stored_session_infos.clone(),
302 self.rebuild_tx.clone(),
303 )),
304 )
305 .await;
306 self.register
307 .add(
308 Type::OfflineMessage,
309 Box::new(OfflineMessageHandler::new(self.cfg.clone(), self.storage_db.clone())),
310 )
311 .await;
312 self.register
313 .add(
314 Type::OfflineInflightMessages,
315 Box::new(OfflineMessageHandler::new(self.cfg.clone(), self.storage_db.clone())),
316 )
317 .await;
318
319 self.load_offline_session_infos().await?;
320
321 Ok(())
322 }
323
324 #[inline]
325 async fn get_config(&self) -> Result<serde_json::Value> {
326 Ok(self.cfg.to_json())
327 }
328
329 #[inline]
330 async fn start(&mut self) -> Result<()> {
331 log::info!("{} start", self.name());
332 *self.scx.extends.session_mgr_mut().await = Box::new(self.session_mgr.clone());
333
334 self.register.start().await;
335 Ok(())
336 }
337
338 #[inline]
339 async fn stop(&mut self) -> Result<bool> {
340 log::warn!("{} stop, if the storage plugin is started, it cannot be stopped", self.name());
341 Ok(false)
342 }
343
344 #[inline]
345 async fn attrs(&self) -> serde_json::Value {
346 async fn stats(storage_db: &DefaultStorageDB) -> (String, String, String, serde_json::Value) {
347 let max_limit = 1000;
348 let mut session_count = 0;
349 let mut storage_db_map = storage_db.clone();
350 {
351 let now = std::time::Instant::now();
352 let iter = storage_db_map.map_iter().await;
353 if let Ok(mut iter) = iter {
354 while let Some(m) = iter.next().await {
355 if let Ok(m) = m {
356 log::debug!("map: {:?}", StoredKey::from(m.name().to_vec()));
357 }
358 session_count += 1;
359 if session_count >= max_limit {
360 break;
361 }
362 }
363 }
364 log::debug!("map_iter cost time: {:?}", now.elapsed());
365 }
366
367 let mut offline_session_count = 0;
368 let mut offline_message_count = 0;
369 let mut storage_db_list = storage_db.clone();
370 {
371 let now = std::time::Instant::now();
372 let iter = storage_db_list.list_iter().await;
373 if let Ok(mut iter) = iter {
374 while let Some(l) = iter.next().await {
375 if let Ok(mut l) = l {
376 log::debug!("list: {:?}", StoredKey::from(l.name().to_vec()));
377 if let Ok(mut l_iter) = l.iter::<OfflineMessageOptionType>().await {
378 while let Some(msg) = l_iter.next().await {
379 if let Ok(Some(_)) = msg {
380 offline_message_count += 1;
381 if offline_message_count >= max_limit {
382 break;
383 }
384 }
385 }
386 }
387 }
388 offline_session_count += 1;
389 if offline_session_count >= max_limit {
390 break;
391 }
392 }
393 }
394 log::debug!("list_iter cost time: {:?}", now.elapsed());
395 }
396 let session_count = if session_count >= max_limit {
397 format!("{session_count}+")
398 } else {
399 format!("{session_count}")
400 };
401 let offline_session_count = if offline_session_count >= max_limit {
402 format!("{offline_session_count}+")
403 } else {
404 format!("{offline_session_count}")
405 };
406 let offline_message_count = if offline_message_count >= max_limit {
407 format!("{offline_message_count}+")
408 } else {
409 format!("{offline_message_count}")
410 };
411
412 let storage_info = storage_db.info().await.unwrap_or_default();
413
414 (session_count, offline_session_count, offline_message_count, storage_info)
415 }
416
417 let (session_count, offline_session_count, offline_message_count, storage_info) =
418 match tokio::time::timeout(Duration::from_secs(1), stats(&self.storage_db)).await {
419 Ok((session_count, offline_session_count, offline_message_count, storage_info)) => {
420 (session_count, offline_session_count, offline_message_count, storage_info)
421 }
422 Err(_) => ("Elapsed".into(), "Elapsed".into(), "Elapsed".into(), serde_json::Value::Null),
423 };
424
425 json!({
426 "session_count": session_count,
427 "offline_session_count": offline_session_count,
428 "offline_message_count": offline_message_count,
429 "storage_info": storage_info
430 })
431 }
432}
433
434struct OfflineMessageHandler {
435 cfg: Arc<PluginConfig>,
436 storage_db: DefaultStorageDB,
437}
438
439impl OfflineMessageHandler {
440 fn new(cfg: Arc<PluginConfig>, storage_db: DefaultStorageDB) -> Self {
441 Self { cfg, storage_db }
442 }
443}
444
445#[async_trait]
446impl Handler for OfflineMessageHandler {
447 async fn hook(&self, param: &Parameter, acc: Option<HookResult>) -> ReturnType {
448 match param {
449 Parameter::OfflineMessage(s, f, p) => {
450 log::debug!(
451 "OfflineMessage storage_type: {:?}, from: {:?}, p: {:?}",
452 self.cfg.storage.typ,
453 f,
454 p
455 );
456 let list_stored_key = make_list_stored_key(s.id.to_string());
457 let storage_db = self.storage_db.clone();
458 let id = s.id.clone();
459 let max_mqueue_len = s.listen_cfg().max_mqueue_len;
460 let p = (*p).clone();
461 let f = f.clone();
462 tokio::spawn(async move {
463 match storage_db.list(list_stored_key.as_ref(), None).await {
464 Ok(offlines_list) => {
465 let res = offlines_list
466 .push_limit::<OfflineMessageOptionType>(
467 &Some((id.client_id.clone(), f, p)),
468 max_mqueue_len,
469 true,
470 )
471 .await;
472 if let Err(e) = res {
473 log::warn!("{id:?} save offline messages error, {e}")
474 }
475 }
476 Err(e) => {
477 log::warn!("{id:?} save offline messages error, {e}")
478 }
479 }
480 });
481 }
482
483 Parameter::OfflineInflightMessages(s, inflight_messages) => {
484 log::debug!(
485 "OfflineInflightMessages storage_type: {:?}, inflight_messages len: {:?}",
486 self.cfg.storage.typ,
487 inflight_messages.len(),
488 );
489 let map_stored_key = make_map_stored_key(s.id.to_string());
490 log::debug!("{:?} map_stored_key: {:?}", s.id, map_stored_key);
491
492 let storage_db = self.storage_db.clone();
493 let inflight_messages = inflight_messages.clone();
494 let id = s.id.clone();
495 tokio::spawn(async move {
496 match storage_db.map(map_stored_key.as_ref(), None).await {
497 Ok(m) => {
498 if let Err(e) = m.insert(INFLIGHT_MESSAGES, &inflight_messages).await {
499 log::warn!("{id:?} save offline inflight messages error, {e}")
500 }
501 }
502 Err(e) => {
503 log::warn!("{id:?} save offline inflight messages error, {e}")
504 }
505 }
506 });
507 }
508
509 _ => {
510 log::error!("unimplemented, {param:?}")
511 }
512 }
513 (true, acc)
514 }
515}
516
517struct StorageHandler {
518 scx: ServerContext,
519 storage_db: DefaultStorageDB,
520 cfg: Arc<PluginConfig>,
521 stored_session_infos: StoredSessionInfos,
522 rebuild_tx: mpsc::Sender<RebuildChanType>,
523}
524
525impl StorageHandler {
526 fn new(
527 scx: ServerContext,
528 storage_db: DefaultStorageDB,
529 cfg: Arc<PluginConfig>,
530 stored_session_infos: StoredSessionInfos,
531 rebuild_tx: mpsc::Sender<RebuildChanType>,
532 ) -> Self {
533 Self { scx, storage_db, cfg, stored_session_infos, rebuild_tx }
534 }
535
536 async fn rebuild_offline_sessions(&self, rebuild_done_tx: oneshot::Sender<()>) {
538 let mut offline_sessions_count = 0;
539 for mut entry in self.stored_session_infos.iter_mut() {
540 let (_, storeds) = entry.pair_mut();
541 if let Some(stored) = storeds.iter_mut().next() {
542 let id = stored.basic.id.clone();
543
544 let listen_cfg =
545 if let Some(listen_cfg) = self.scx.listen_cfgs.get(&id.lid).map(|c| c.value().clone()) {
546 listen_cfg
547 } else {
548 log::warn!("tcp listener config is not found, local addr is {:?}", id.local_addr);
549 continue;
550 };
551
552 log::info!("{id:?} listen_cfg: {listen_cfg:?}");
553
554 let fitter = self.scx.extends.fitter_mgr().await.create(
556 stored.basic.conn_info.clone(),
557 id.clone(),
558 listen_cfg.clone(),
559 );
560
561 let session_expiry_interval = session_expiry_interval(
563 fitter.as_ref(),
564 stored.disconnect_info.as_ref(),
565 stored.last_time,
566 )
567 .await;
568 log::debug!("{id:?} session_expiry_interval: {session_expiry_interval:?}");
569 if session_expiry_interval <= 0 {
570 log::debug!(
571 "{:?} session is expiry, {:?}, id_key: {:?}, {:?}, {:?}",
572 id,
573 session_expiry_interval,
574 stored.id_key,
575 make_map_stored_key(stored.id_key.as_ref()),
576 make_list_stored_key(stored.id_key.as_ref())
577 );
578 let storage_db = self.storage_db.clone();
579 if let Err(e) = storage_db.map_remove(make_map_stored_key(stored.id_key.as_ref())).await {
580 log::warn!("{id:?} remove map error, {e:?}");
581 }
582 if let Err(e) = storage_db.list_remove(make_list_stored_key(stored.id_key.as_ref())).await
583 {
584 log::warn!("{id:?} remove list error, {e:?}");
585 }
586 continue;
588 }
589 offline_sessions_count += 1;
590
591 if stored.disconnect_info.is_none() {
592 stored.disconnect_info = Some(DisconnectInfo::new(stored.last_time));
593 }
594
595 let max_inflight = fitter.max_inflight();
596 let max_mqueue_len = fitter.max_mqueue_len();
597 let subs = stored.subs.take().map(SessionSubs::from).unwrap_or_else(SessionSubs::new);
598
599 let session = match Session::new(
600 id.clone(),
601 self.scx.clone(),
602 max_mqueue_len,
603 listen_cfg,
604 fitter,
605 None,
606 max_inflight,
607 stored.basic.created_at,
608 stored.basic.conn_info.clone(),
609 false,
610 false,
611 false,
612 stored.basic.connected_at,
613 subs,
614 stored.disconnect_info.take(),
615 None,
616 )
617 .await
618 {
619 Ok(s) => s,
620 Err(e) => {
621 log::warn!("rebuild session offline message error, create session error, {e:?}");
622 continue;
623 }
624 };
625
626 let deliver_queue = session.deliver_queue();
627 for item in stored.offline_messages.drain(..) {
628 if let Err((f, p)) = deliver_queue.push(item) {
629 log::warn!("rebuild session offline message error, deliver queue is full, from: {f:?}, publish: {p:?}");
630 }
631 }
632
633 let out_inflight = session.out_inflight();
634 for item in stored.inflight_messages.drain(..) {
635 out_inflight.write().await.push_back(item);
636 }
637
638 if let Err(e) = self
639 .rebuild_tx
640 .clone()
641 .send(RebuildChanType::Session(
642 session,
643 Duration::from_millis(session_expiry_interval as u64),
644 ))
645 .await
646 {
647 log::error!("rebuild offline sessions error, {e:?}");
648 }
649 }
650 }
651 log::info!("offline_sessions_count: {offline_sessions_count}");
652 let _ = self.rebuild_tx.clone().send(RebuildChanType::Done(rebuild_done_tx)).await;
653 }
654}
655
656#[async_trait]
657impl Handler for StorageHandler {
658 async fn hook(&self, param: &Parameter, acc: Option<HookResult>) -> ReturnType {
659 match param {
660 Parameter::BeforeStartup => {
661 log::info!(
662 "BeforeStartup storage_type: {:?}, stored_session_infos len: {}",
663 self.cfg.storage.typ,
664 self.stored_session_infos.len()
665 );
666 let (rebuild_done_tx, rebuild_done_rx) = oneshot::channel::<()>();
667 self.rebuild_offline_sessions(rebuild_done_tx).await;
668 let _ = rebuild_done_rx.await;
669 }
670 _ => {
671 log::error!("unimplemented, {param:?}")
672 }
673 }
674 (true, acc)
675 }
676}
677
678#[inline]
679async fn session_expiry_interval(
680 fitter: &dyn Fitter,
681 disconnect_info: Option<&DisconnectInfo>,
682 last_time: TimestampMillis,
683) -> TimestampMillis {
684 let disconnected_at = disconnect_info.map(|d| d.disconnected_at).unwrap_or_default();
685 let disconnected_at = if disconnected_at <= 0 { last_time } else { disconnected_at };
686 fitter.session_expiry_interval(disconnect_info.and_then(|d| d.mqtt_disconnect.as_ref())).as_millis()
687 as i64
688 - (timestamp_millis() - disconnected_at)
689}
690
691#[inline]
692pub(crate) fn make_map_stored_key<T: AsRef<[u8]>>(id: T) -> StoredKey {
693 let mut key = Vec::from("map-");
694 key.extend_from_slice(id.as_ref());
695 Bytes::from(key)
696}
697
698#[inline]
699pub(crate) fn map_stored_key_to_id_bytes(stored_key: &[u8]) -> &[u8] {
700 if stored_key.starts_with(b"map-") {
701 stored_key[4..].as_ref()
702 } else {
703 stored_key
704 }
705}
706
707#[inline]
708pub(crate) fn make_list_stored_key<T: AsRef<[u8]>>(id: T) -> StoredKey {
709 let mut key = Vec::from("list-");
710 key.extend_from_slice(id.as_ref());
711 Bytes::from(key)
712}
713
714#[inline]
715pub(crate) fn list_stored_key_to_id_bytes(stored_key: &[u8]) -> &[u8] {
716 if stored_key.starts_with(b"list-") {
717 stored_key[5..].as_ref()
718 } else {
719 stored_key
720 }
721}