1
0

list.rs 58 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007
  1. //! # List command handlers
  2. use crate::{
  3. check_arg,
  4. connection::{Connection, ConnectionStatus, UnblockReason},
  5. db::utils::far_future,
  6. error::Error,
  7. try_get_arg, try_get_arg_str,
  8. value::bytes_to_number,
  9. value::checksum,
  10. value::Value,
  11. };
  12. use bytes::Bytes;
  13. use std::collections::VecDeque;
  14. use tokio::time::{sleep, Duration, Instant};
  15. #[allow(clippy::needless_range_loop)]
  16. /// Removes an element from a list
  17. #[inline]
  18. fn remove_element(
  19. conn: &Connection,
  20. key: &Bytes,
  21. limit: Option<usize>,
  22. front: bool,
  23. ) -> Result<Value, Error> {
  24. let db = conn.db();
  25. let mut new_len = 0;
  26. let result = db.get_map_or(
  27. key,
  28. |v| match v {
  29. Value::List(x) => {
  30. let mut x = x.write();
  31. let limit = if let Some(limit) = limit {
  32. limit
  33. } else {
  34. // Return a single element
  35. let ret = Ok((if front { x.pop_front() } else { x.pop_back() })
  36. .map_or(Value::Null, |x| x.clone_value()));
  37. new_len = x.len();
  38. return ret;
  39. };
  40. let mut ret = vec![None; limit];
  41. for i in 0..limit {
  42. if front {
  43. ret[i] = x.pop_front();
  44. } else {
  45. ret[i] = x.pop_back();
  46. }
  47. }
  48. new_len = x.len();
  49. Ok(ret
  50. .iter()
  51. .flatten()
  52. .map(|m| m.clone_value())
  53. .collect::<Vec<Value>>()
  54. .into())
  55. }
  56. _ => Err(Error::WrongType),
  57. },
  58. || Ok(Value::Null),
  59. )?;
  60. if new_len == 0 {
  61. let _ = db.del(&[key.clone()]);
  62. } else {
  63. db.bump_version(key);
  64. }
  65. Ok(result)
  66. }
  67. #[inline]
  68. /// Handles the timeout/sleep logic for all blocking commands.
  69. async fn handle_timeout(conn: &Connection, timeout: Option<Instant>) -> Result<bool, Error> {
  70. if let Some(timeout) = timeout {
  71. if Instant::now() >= timeout {
  72. conn.unblock(UnblockReason::Timeout);
  73. return Ok(true);
  74. }
  75. }
  76. if let Some(reason) = conn.has_been_unblocked_externally() {
  77. match reason {
  78. UnblockReason::Error => Err(Error::UnblockByError),
  79. _ => Ok(true),
  80. }
  81. } else {
  82. sleep(Duration::from_millis(100)).await;
  83. Ok(false)
  84. }
  85. }
  86. /// Parses timeout and returns an instant or none if it should wait forever.
  87. #[inline]
  88. fn parse_timeout(arg: &Bytes) -> Result<Option<Instant>, Error> {
  89. let raw_timeout = bytes_to_number::<f64>(arg)?;
  90. if raw_timeout < 0f64 {
  91. return Err(Error::NegativeNumber("timeout".to_owned()));
  92. }
  93. if raw_timeout == 0.0 {
  94. return Ok(None);
  95. }
  96. Ok(Some(
  97. Instant::now()
  98. .checked_add(Duration::from_millis(
  99. (raw_timeout * 1_000f64).round() as u64
  100. ))
  101. .unwrap_or_else(far_future),
  102. ))
  103. }
  104. /// BLPOP is a blocking list pop primitive. It is the blocking version of LPOP because it blocks
  105. /// the connection when there are no elements to pop from any of the given lists. An element is
  106. /// popped from the head of the first list that is non-empty, with the given keys being checked in
  107. /// the order that they are given.
  108. pub async fn blpop(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  109. let blpop_task = |conn: &Connection, args: &[Bytes]| -> Result<Value, Error> {
  110. for key in (1..args.len() - 1) {
  111. let key = &args[key];
  112. match remove_element(&conn, key, None, true)? {
  113. Value::Null => (),
  114. n => return Ok(vec![Value::new(&key), n].into()),
  115. };
  116. }
  117. Ok(Value::Null)
  118. };
  119. if conn.is_executing_tx() {
  120. return blpop_task(conn, args);
  121. }
  122. let timeout = parse_timeout(&args[args.len() - 1])?;
  123. let conn = conn.clone();
  124. let args = args.to_vec();
  125. conn.block();
  126. tokio::spawn(async move {
  127. loop {
  128. match blpop_task(&conn, &args) {
  129. Ok(Value::Null) => {}
  130. Ok(x) => {
  131. conn.append_response(x);
  132. conn.unblock(UnblockReason::Finished);
  133. break;
  134. }
  135. Err(x) => {
  136. conn.append_response(x.into());
  137. conn.unblock(UnblockReason::Finished);
  138. break;
  139. }
  140. }
  141. match handle_timeout(&conn, timeout).await {
  142. Ok(true) => {
  143. conn.append_response(Value::Null);
  144. break;
  145. }
  146. Err(x) => {
  147. conn.append_response(x.into());
  148. break;
  149. }
  150. _ => {}
  151. }
  152. }
  153. });
  154. Ok(Value::Ignore)
  155. }
  156. /// BLMOVE is the blocking variant of LMOVE. When source contains elements, this
  157. /// command behaves exactly like LMOVE. When used inside a MULTI/EXEC block,
  158. /// this command behaves exactly like LMOVE. When source is empty, Redis will
  159. /// block the connection until another client pushes to it or until timeout (a
  160. /// double value specifying the maximum number of seconds to block) is reached.
  161. /// A timeout of zero can be used to block indefinitely.
  162. ///
  163. /// This command comes in place of the now deprecated BRPOPLPUSH. Doing BLMOVE
  164. /// RIGHT LEFT is equivalent.
  165. ///
  166. /// See LMOVE for more information.
  167. pub async fn blmove(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  168. if conn.is_executing_tx() {
  169. return lmove(&conn, &args).await;
  170. }
  171. let timeout = parse_timeout(&args[5])?;
  172. conn.block();
  173. let conn = conn.clone();
  174. let args = args.to_vec();
  175. tokio::spawn(async move {
  176. loop {
  177. match lmove(&conn, &args).await {
  178. Ok(Value::Null) => (),
  179. Ok(n) => {
  180. conn.append_response(n);
  181. conn.unblock(UnblockReason::Finished);
  182. break;
  183. }
  184. Err(x) => {
  185. conn.append_response(x.into());
  186. conn.unblock(UnblockReason::Finished);
  187. break;
  188. }
  189. };
  190. match handle_timeout(&conn, timeout).await {
  191. Ok(true) => {
  192. conn.append_response(Value::Null);
  193. break;
  194. }
  195. Err(x) => {
  196. conn.append_response(x.into());
  197. break;
  198. }
  199. _ => {}
  200. }
  201. }
  202. });
  203. Ok(Value::Ignore)
  204. }
  205. /// BRPOPLPUSH is the blocking variant of RPOPLPUSH. When source contains
  206. /// elements, this command behaves exactly like RPOPLPUSH. When used inside a
  207. /// MULTI/EXEC block, this command behaves exactly like RPOPLPUSH. When source
  208. /// is empty, Redis will block the connection until another client pushes to it
  209. /// or until timeout is reached. A timeout of zero can be used to block
  210. /// indefinitely.
  211. pub async fn brpoplpush(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  212. blmove(
  213. conn,
  214. &[
  215. "blmove".into(),
  216. args[1].clone(),
  217. args[2].clone(),
  218. "RIGHT".into(),
  219. "LEFT".into(),
  220. args[3].clone(),
  221. ],
  222. )
  223. .await
  224. }
  225. /// BRPOP is a blocking list pop primitive. It is the blocking version of RPOP because it blocks
  226. /// the connection when there are no elements to pop from any of the given lists. An element is
  227. /// popped from the tail of the first list that is non-empty, with the given keys being checked in
  228. /// the order that they are given.
  229. pub async fn brpop(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  230. let brpop_task = |conn: &Connection, args: &[Bytes]| -> Result<Value, Error> {
  231. for key in (1..args.len() - 1) {
  232. let key = &args[key];
  233. match remove_element(&conn, key, None, false)? {
  234. Value::Null => (),
  235. n => return Ok(vec![Value::new(&key), n].into()),
  236. };
  237. }
  238. Ok(Value::Null)
  239. };
  240. if conn.is_executing_tx() {
  241. return brpop_task(conn, args);
  242. }
  243. let timeout = parse_timeout(&args[args.len() - 1])?;
  244. let conn = conn.clone();
  245. let args = args.to_vec();
  246. conn.block();
  247. tokio::spawn(async move {
  248. loop {
  249. match brpop_task(&conn, &args) {
  250. Ok(Value::Null) => {}
  251. Ok(x) => {
  252. conn.append_response(x);
  253. conn.unblock(UnblockReason::Finished);
  254. break;
  255. }
  256. Err(x) => {
  257. conn.append_response(x.into());
  258. conn.unblock(UnblockReason::Finished);
  259. break;
  260. }
  261. }
  262. match handle_timeout(&conn, timeout).await {
  263. Ok(true) => {
  264. conn.append_response(Value::Null);
  265. break;
  266. }
  267. Err(x) => {
  268. conn.append_response(x.into());
  269. break;
  270. }
  271. _ => {}
  272. }
  273. }
  274. });
  275. Ok(Value::Ignore)
  276. }
  277. /// Returns the element at index index in the list stored at key. The index is zero-based, so 0
  278. /// means the first element, 1 the second element and so on. Negative indices can be used to
  279. /// designate elements starting at the tail of the list. Here, -1 means the last element, -2 means
  280. /// the penultimate and so forth.
  281. pub async fn lindex(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  282. conn.db().get_map_or(
  283. &args[1],
  284. |v| match v {
  285. Value::List(x) => {
  286. let mut index: i64 = bytes_to_number(&args[2])?;
  287. let x = x.read();
  288. let index = if index < 0 {
  289. x.len()
  290. .checked_sub((index * -1) as usize)
  291. .unwrap_or(x.len())
  292. } else {
  293. index as usize
  294. };
  295. Ok(x.get(index).map_or(Value::Null, |x| x.clone_value()))
  296. }
  297. _ => Err(Error::WrongType),
  298. },
  299. || Ok(Value::Null),
  300. )
  301. }
  302. /// Inserts element in the list stored at key either before or after the reference value pivot.
  303. ///
  304. /// When key does not exist, it is considered an empty list and no operation is performed.
  305. pub async fn linsert(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  306. let is_before = if check_arg!(args, 2, "BEFORE") {
  307. true
  308. } else if check_arg!(args, 2, "AFTER") {
  309. false
  310. } else {
  311. return Err(Error::Syntax);
  312. };
  313. let result = conn.db().get_map_or(
  314. &args[1],
  315. |v| match v {
  316. Value::List(x) => {
  317. let pivot = checksum::Ref::new(&args[3]);
  318. let mut x = x.write();
  319. let mut found = false;
  320. for (key, val) in x.iter().enumerate() {
  321. if *val == pivot {
  322. let id = if is_before { key } else { key + 1 };
  323. let value = checksum::Value::new(args[4].clone());
  324. if id > x.len() {
  325. x.push_back(value);
  326. } else {
  327. x.insert(id as usize, value);
  328. }
  329. found = true;
  330. break;
  331. }
  332. }
  333. if found {
  334. Ok(x.len().into())
  335. } else {
  336. Ok((-1).into())
  337. }
  338. }
  339. _ => Err(Error::WrongType),
  340. },
  341. || Ok(0.into()),
  342. )?;
  343. conn.db().bump_version(&args[1]);
  344. Ok(result)
  345. }
  346. /// Returns the length of the list stored at key. If key does not exist, it is interpreted as an
  347. /// empty list and 0 is returned. An error is returned when the value stored at key is not a list.
  348. pub async fn llen(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  349. conn.db().get_map_or(
  350. &args[1],
  351. |v| match v {
  352. Value::List(x) => Ok(x.read().len().into()),
  353. _ => Err(Error::WrongType),
  354. },
  355. || Ok(0.into()),
  356. )
  357. }
  358. /// Atomically returns and removes the first/last element (head/tail depending on the wherefrom
  359. /// argument) of the list stored at source, and pushes the element at the first/last element
  360. /// (head/tail depending on the whereto argument) of the list stored at destination.
  361. pub async fn lmove(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  362. let source_is_left = if check_arg!(args, 3, "LEFT") {
  363. true
  364. } else if check_arg!(args, 3, "RIGHT") {
  365. false
  366. } else {
  367. return Err(Error::Syntax);
  368. };
  369. let target_is_left = if check_arg!(args, 4, "LEFT") {
  370. true
  371. } else if check_arg!(args, 4, "RIGHT") {
  372. false
  373. } else {
  374. return Err(Error::Syntax);
  375. };
  376. let db = conn.db();
  377. /// Lock keys to alter exclusively
  378. db.lock_keys(&args[1..=2]);
  379. let result = db.get_map_or(
  380. &args[1],
  381. |v| match v {
  382. Value::List(source) => conn.db().get_map_or(
  383. &args[2],
  384. |v| match v {
  385. Value::List(target) => {
  386. let element = if source_is_left {
  387. source.write().pop_front()
  388. } else {
  389. source.write().pop_back()
  390. };
  391. if let Some(element) = element {
  392. let ret = element.clone_value();
  393. if target_is_left {
  394. target.write().push_front(element);
  395. } else {
  396. target.write().push_back(element);
  397. }
  398. Ok(ret)
  399. } else {
  400. Ok(Value::Null)
  401. }
  402. }
  403. _ => Err(Error::WrongType),
  404. },
  405. || {
  406. let mut source = source.write();
  407. let element = if source_is_left {
  408. source.pop_front()
  409. } else {
  410. source.pop_back()
  411. };
  412. if let Some(element) = element {
  413. let ret = element.clone_value();
  414. let mut h = VecDeque::new();
  415. h.push_front(element);
  416. conn.db().set(&args[2], h.into(), None);
  417. Ok(ret)
  418. } else {
  419. Ok(Value::Null)
  420. }
  421. },
  422. ),
  423. _ => Err(Error::WrongType),
  424. },
  425. || Ok(Value::Null),
  426. );
  427. /// release the lock on keys
  428. db.unlock_keys(&args[1..=2]);
  429. if result != Ok(Value::Null) {
  430. conn.db().bump_version(&args[1]);
  431. conn.db().bump_version(&args[2]);
  432. }
  433. result
  434. }
  435. /// Removes and returns the first elements of the list stored at key.
  436. ///
  437. /// By default, the command pops a single element from the beginning of the list. When provided
  438. /// with the optional count argument, the reply will consist of up to count elements, depending on
  439. /// the list's length.
  440. pub async fn lpop(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  441. let count = match args.get(2) {
  442. Some(v) => Some(bytes_to_number(&v)?),
  443. None => None,
  444. };
  445. remove_element(conn, &args[1], count, true)
  446. }
  447. /// The command returns the index of matching elements inside a Redis list. By default, when no
  448. /// options are given, it will scan the list from head to tail, looking for the first match of
  449. /// "element". If the element is found, its index (the zero-based position in the list) is
  450. /// returned. Otherwise, if no match is found, nil is returned.
  451. pub async fn lpos(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  452. let element = checksum::Ref::new(&args[2]);
  453. let mut rank = None;
  454. let mut count = None;
  455. let mut max_len = None;
  456. let mut index = 3;
  457. loop {
  458. if args.len() <= index {
  459. break;
  460. }
  461. let next = try_get_arg!(args, index + 1);
  462. match try_get_arg_str!(args, index).to_uppercase().as_str() {
  463. "RANK" => rank = Some(bytes_to_number::<i64>(&next)?),
  464. "COUNT" => count = Some(bytes_to_number::<usize>(&next)?),
  465. "MAXLEN" => max_len = Some(bytes_to_number::<usize>(&next)?),
  466. _ => return Err(Error::Syntax),
  467. }
  468. index += 2;
  469. }
  470. let (must_reverse, rank) = if let Some(rank) = rank {
  471. if rank == 0 {
  472. return Err(Error::InvalidRank("RANK".to_owned()));
  473. }
  474. if rank < 0 {
  475. (true, Some((rank * -1) as usize))
  476. } else {
  477. (false, Some(rank as usize))
  478. }
  479. } else {
  480. (false, None)
  481. };
  482. let max_len = max_len.unwrap_or_default();
  483. conn.db().get_map_or(
  484. &args[1],
  485. |v| match v {
  486. Value::List(x) => {
  487. let x = x.read();
  488. let mut result: Vec<Value> = vec![];
  489. let mut values = x
  490. .iter()
  491. .enumerate()
  492. .collect::<Vec<(usize, &checksum::Value)>>();
  493. if must_reverse {
  494. values.reverse();
  495. }
  496. let mut checks = 1;
  497. for (id, val) in values.iter() {
  498. if **val == element {
  499. // Match!
  500. if let Some(count) = count {
  501. result.push((*id).into());
  502. if result.len() == count && count != 0 && rank.is_none() {
  503. // There is no point in keep looping. No RANK provided, COUNT is not 0
  504. // therefore we can return the vector of result as IS
  505. return Ok(result.into());
  506. }
  507. } else if let Some(rank) = rank {
  508. result.push((*id).into());
  509. if result.len() == rank {
  510. return Ok((*id).into());
  511. }
  512. } else {
  513. // return first match!
  514. return Ok((*id).into());
  515. }
  516. }
  517. if checks == max_len {
  518. break;
  519. }
  520. checks += 1;
  521. }
  522. if let Some(rank) = rank {
  523. let rank = rank - 1;
  524. let result = if rank < result.len() {
  525. (&result[rank..]).to_vec()
  526. } else {
  527. vec![]
  528. };
  529. return Ok(if let Some(count) = count {
  530. if count > 0 && count < result.len() {
  531. (&result[0..count]).to_vec().into()
  532. } else {
  533. result.to_vec().into()
  534. }
  535. } else {
  536. result
  537. .to_vec()
  538. .get(0)
  539. .map(|c| c.clone())
  540. .unwrap_or_default()
  541. });
  542. }
  543. if count.is_some() {
  544. Ok(result.into())
  545. } else {
  546. Ok(Value::Null)
  547. }
  548. }
  549. _ => Err(Error::WrongType),
  550. },
  551. || {
  552. Ok(if count.is_some() {
  553. Value::Array(vec![])
  554. } else {
  555. Value::Null
  556. })
  557. },
  558. )
  559. }
  560. /// Insert all the specified values at the head of the list stored at key. If key does not exist,
  561. /// it is created as empty list before performing the push operations. When key holds a value that
  562. /// is not a list, an error is returned.
  563. ///
  564. /// It is possible to push multiple elements using a single command call just specifying multiple
  565. /// arguments at the end of the command. Elements are inserted one after the other to the head of
  566. /// the list, from the leftmost element to the rightmost element. So for instance the command LPUSH
  567. /// mylist a b c will result into a list containing c as first element, b as second element and a
  568. /// as third element.
  569. pub async fn lpush(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  570. let is_push_x = check_arg!(args, 0, "LPUSHX");
  571. let result = conn.db().get_map_or(
  572. &args[1],
  573. |v| match v {
  574. Value::List(x) => {
  575. let mut x = x.write();
  576. for val in args.iter().skip(2) {
  577. x.push_front(checksum::Value::new(val.clone()));
  578. }
  579. Ok(x.len().into())
  580. }
  581. _ => Err(Error::WrongType),
  582. },
  583. || {
  584. if is_push_x {
  585. return Ok(0.into());
  586. }
  587. let mut h = VecDeque::new();
  588. for val in args.iter().skip(2) {
  589. h.push_front(checksum::Value::new(val.clone()));
  590. }
  591. let len = h.len();
  592. conn.db().set(&args[1], h.into(), None);
  593. Ok(len.into())
  594. },
  595. )?;
  596. conn.db().bump_version(&args[1]);
  597. Ok(result)
  598. }
  599. /// Returns the specified elements of the list stored at key. The offsets start and stop are
  600. /// zero-based indexes, with 0 being the first element of the list (the head of the list), 1 being
  601. /// the next element and so on.
  602. ///
  603. /// These offsets can also be negative numbers indicating offsets starting at the end of the list.
  604. /// For example, -1 is the last element of the list, -2 the penultimate, and so on.
  605. pub async fn lrange(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  606. conn.db().get_map_or(
  607. &args[1],
  608. |v| match v {
  609. Value::List(x) => {
  610. let start: i64 = bytes_to_number(&args[2])?;
  611. let end: i64 = bytes_to_number(&args[3])?;
  612. let mut ret = vec![];
  613. let x = x.read();
  614. let start = if start < 0 {
  615. x.len()
  616. .checked_sub((start * -1) as usize)
  617. .unwrap_or_default()
  618. } else {
  619. (start as usize)
  620. };
  621. let end = if end < 0 {
  622. if let Some(x) = x.len().checked_sub((end * -1) as usize) {
  623. x
  624. } else {
  625. return Ok(Value::Array((vec![])));
  626. }
  627. } else {
  628. end as usize
  629. };
  630. for (i, val) in x.iter().enumerate().skip(start) {
  631. if i > end {
  632. break;
  633. }
  634. ret.push(val.clone_value());
  635. }
  636. Ok(ret.into())
  637. }
  638. _ => Err(Error::WrongType),
  639. },
  640. || Ok(Value::Array(vec![])),
  641. )
  642. }
  643. /// Removes the first count occurrences of elements equal to element from the list stored at key
  644. pub async fn lrem(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  645. let result = conn.db().get_map_or(
  646. &args[1],
  647. |v| match v {
  648. Value::List(x) => {
  649. let element = checksum::Ref::new(&args[3]);
  650. let limit: i64 = bytes_to_number(&args[2])?;
  651. let mut x = x.write();
  652. let (is_reverse, limit) = if limit < 0 {
  653. (true, -limit)
  654. } else {
  655. (false, limit)
  656. };
  657. let mut keep = vec![true; x.len()];
  658. let mut removed = 0;
  659. let len = x.len();
  660. for i in 0..len {
  661. let i = if is_reverse { len - 1 - i } else { i };
  662. if let Some(value) = x.get(i) {
  663. if *value == element {
  664. keep[i] = false;
  665. removed += 1;
  666. if removed == limit {
  667. break;
  668. }
  669. }
  670. }
  671. }
  672. let mut i = 0;
  673. x.retain(|_| {
  674. i += 1;
  675. keep[i - 1]
  676. });
  677. Ok(removed.into())
  678. }
  679. _ => Err(Error::WrongType),
  680. },
  681. || Ok(0.into()),
  682. )?;
  683. conn.db().bump_version(&args[1]);
  684. Ok(result)
  685. }
  686. /// Sets the list element at index to element. For more information on the index argument, see
  687. /// LINDEX.
  688. ///
  689. /// An error is returned for out of range indexes.
  690. pub async fn lset(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  691. let result = conn.db().get_map_or(
  692. &args[1],
  693. |v| match v {
  694. Value::List(x) => {
  695. let mut index: i64 = bytes_to_number(&args[2])?;
  696. let mut x = x.write();
  697. if index < 0 {
  698. index += x.len() as i64;
  699. }
  700. if let Some(x) = x.get_mut(index as usize) {
  701. *x = checksum::Value::new(args[3].clone());
  702. Ok(Value::Ok)
  703. } else {
  704. Err(Error::OutOfRange)
  705. }
  706. }
  707. _ => Err(Error::WrongType),
  708. },
  709. || Err(Error::NotFound),
  710. )?;
  711. conn.db().bump_version(&args[1]);
  712. Ok(result)
  713. }
  714. /// Trim an existing list so that it will contain only the specified range of elements specified.
  715. /// Both start and stop are zero-based indexes, where 0 is the first element of the list (the
  716. /// head), 1 the next element and so on.
  717. pub async fn ltrim(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  718. let result = conn.db().get_map_or(
  719. &args[1],
  720. |v| match v {
  721. Value::List(x) => {
  722. let mut start: i64 = bytes_to_number(&args[2])?;
  723. let mut end: i64 = bytes_to_number(&args[3])?;
  724. let mut x = x.write();
  725. if start < 0 {
  726. start += x.len() as i64;
  727. }
  728. if end < 0 {
  729. end += x.len() as i64;
  730. }
  731. let mut i = 0;
  732. x.retain(|_| {
  733. let retain = i >= start && i <= end;
  734. i += 1;
  735. retain
  736. });
  737. Ok(Value::Ok)
  738. }
  739. _ => Err(Error::WrongType),
  740. },
  741. || Ok(Value::Ok),
  742. )?;
  743. conn.db().bump_version(&args[1]);
  744. Ok(result)
  745. }
  746. /// Removes and returns the last elements of the list stored at key.
  747. ///
  748. /// By default, the command pops a single element from the end of the list. When provided with the
  749. /// optional count argument, the reply will consist of up to count elements, depending on the
  750. /// list's length.
  751. pub async fn rpop(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  752. let count = match args.get(2) {
  753. Some(v) => Some(bytes_to_number(&v)?),
  754. None => None,
  755. };
  756. remove_element(conn, &args[1], count, false)
  757. }
  758. /// Atomically returns and removes the last element (tail) of the list stored at source, and pushes
  759. /// the element at the first element (head) of the list stored at destination.
  760. pub async fn rpoplpush(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  761. lmove(
  762. conn,
  763. &[
  764. "lmove".into(),
  765. args[1].clone(),
  766. args[2].clone(),
  767. "RIGHT".into(),
  768. "LEFT".into(),
  769. ],
  770. )
  771. .await
  772. }
  773. /// Insert all the specified values at the tail of the list stored at key. If key does not exist,
  774. /// it is created as empty list before performing the push operation. When key holds a value that
  775. /// is not a list, an error is returned.
  776. pub async fn rpush(conn: &Connection, args: &[Bytes]) -> Result<Value, Error> {
  777. let is_push_x = check_arg!(args, 0, "RPUSHX");
  778. let result = conn.db().get_map_or(
  779. &args[1],
  780. |v| match v {
  781. Value::List(x) => {
  782. let mut x = x.write();
  783. for val in args.iter().skip(2) {
  784. x.push_back(checksum::Value::new(val.clone()));
  785. }
  786. Ok(x.len().into())
  787. }
  788. _ => Err(Error::WrongType),
  789. },
  790. || {
  791. if is_push_x {
  792. return Ok(0.into());
  793. }
  794. let mut h = VecDeque::new();
  795. for val in args.iter().skip(2) {
  796. h.push_back(checksum::Value::new(val.clone()));
  797. }
  798. let len = h.len();
  799. conn.db().set(&args[1], h.into(), None);
  800. Ok(len.into())
  801. },
  802. )?;
  803. conn.db().bump_version(&args[1]);
  804. Ok(result)
  805. }
  806. #[cfg(test)]
  807. mod test {
  808. use crate::{
  809. cmd::test::{create_connection, create_connection_and_pubsub, run_command},
  810. error::Error,
  811. value::Value,
  812. };
  813. use tokio::time::{sleep, Duration, Instant};
  814. #[tokio::test]
  815. async fn blpop_no_waiting() {
  816. let (mut recv, c) = create_connection_and_pubsub();
  817. assert_eq!(
  818. Ok(Value::Integer(5)),
  819. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await,
  820. );
  821. assert_eq!(
  822. Ok(Value::Ignore),
  823. run_command(&c, &["blpop", "foo", "1"]).await
  824. );
  825. assert_eq!(
  826. Some(Value::Array(vec![
  827. Value::Blob("foo".into()),
  828. Value::Blob("5".into()),
  829. ])),
  830. recv.recv().await
  831. );
  832. }
  833. #[tokio::test]
  834. async fn blpop_timeout() {
  835. let (mut recv, c) = create_connection_and_pubsub();
  836. let x = Instant::now();
  837. assert_eq!(
  838. Ok(Value::Ignore),
  839. run_command(&c, &["blpop", "foobar", "1"]).await
  840. );
  841. assert_eq!(Some(Value::Null), recv.recv().await,);
  842. assert!(Instant::now() - x >= Duration::from_millis(1000));
  843. }
  844. #[tokio::test]
  845. async fn blpop_wait_insert() {
  846. let (mut recv, c) = create_connection_and_pubsub();
  847. let x = Instant::now();
  848. // Query command that will block connection until some data is inserted
  849. // to foobar, foo, bar or the 5 seconds timeout happens.
  850. //
  851. // We are issuing the command, sleeping a little bit then adding data to
  852. // foobar, before actually waiting on the result.
  853. assert_eq!(
  854. Ok(Value::Ignore),
  855. run_command(&c, &["blpop", "foobar", "foo", "bar", "5"]).await
  856. );
  857. // Sleep 1 second before inserting new data
  858. sleep(Duration::from_millis(1000)).await;
  859. assert_eq!(
  860. Ok(Value::Integer(5)),
  861. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await,
  862. );
  863. // Read the output of the first blpop command now.
  864. assert_eq!(
  865. Some(Value::Array(vec![
  866. Value::Blob("foo".into()),
  867. Value::Blob("5".into()),
  868. ])),
  869. recv.recv().await,
  870. );
  871. assert!(Instant::now() - x > Duration::from_millis(1000));
  872. assert!(Instant::now() - x < Duration::from_millis(5000));
  873. }
  874. #[tokio::test]
  875. async fn lrem_1() {
  876. let c = create_connection();
  877. assert_eq!(
  878. Ok(Value::Integer(5)),
  879. run_command(
  880. &c,
  881. &["rpush", "mylist", "hello", "hello", "world", "hello", "hello"]
  882. )
  883. .await
  884. );
  885. assert_eq!(
  886. Ok(Value::Integer(3)),
  887. run_command(&c, &["lrem", "mylist", "3", "hello"]).await
  888. );
  889. assert_eq!(
  890. Ok(Value::Array(vec![
  891. Value::Blob("world".into()),
  892. Value::Blob("hello".into()),
  893. ])),
  894. run_command(&c, &["lrange", "mylist", "0", "-1"]).await
  895. );
  896. }
  897. #[tokio::test]
  898. async fn lrem_2() {
  899. let c = create_connection();
  900. assert_eq!(
  901. Ok(Value::Integer(5)),
  902. run_command(
  903. &c,
  904. &["rpush", "mylist", "hello", "hello", "world", "hello", "hello"]
  905. )
  906. .await
  907. );
  908. assert_eq!(
  909. Ok(Value::Integer(2)),
  910. run_command(&c, &["lrem", "mylist", "-2", "hello"]).await
  911. );
  912. assert_eq!(
  913. Ok(Value::Array(vec![
  914. Value::Blob("hello".into()),
  915. Value::Blob("hello".into()),
  916. Value::Blob("world".into()),
  917. ])),
  918. run_command(&c, &["lrange", "mylist", "0", "-1"]).await
  919. );
  920. assert_eq!(
  921. Ok(Value::Integer(1)),
  922. run_command(&c, &["lrem", "mylist", "1", "hello"]).await
  923. );
  924. assert_eq!(
  925. Ok(Value::Array(vec![
  926. Value::Blob("hello".into()),
  927. Value::Blob("world".into()),
  928. ])),
  929. run_command(&c, &["lrange", "mylist", "0", "-1"]).await
  930. );
  931. }
  932. #[tokio::test]
  933. async fn lrem_3() {
  934. let c = create_connection();
  935. assert_eq!(
  936. Ok(Value::Integer(5)),
  937. run_command(
  938. &c,
  939. &["rpush", "mylist", "hello", "hello", "world", "hello", "hello"]
  940. )
  941. .await
  942. );
  943. assert_eq!(
  944. Ok(Value::Integer(4)),
  945. run_command(&c, &["lrem", "mylist", "-100", "hello"]).await
  946. );
  947. assert_eq!(
  948. Ok(Value::Array(vec![Value::Blob("world".into()),])),
  949. run_command(&c, &["lrange", "mylist", "0", "-1"]).await
  950. );
  951. }
  952. #[tokio::test]
  953. async fn lrem_4() {
  954. let c = create_connection();
  955. assert_eq!(
  956. Ok(Value::Integer(5)),
  957. run_command(
  958. &c,
  959. &["rpush", "mylist", "hello", "hello", "world", "hello", "hello"]
  960. )
  961. .await
  962. );
  963. assert_eq!(
  964. Ok(Value::Integer(4)),
  965. run_command(&c, &["lrem", "mylist", "100", "hello"]).await
  966. );
  967. assert_eq!(
  968. Ok(Value::Array(vec![Value::Blob("world".into()),])),
  969. run_command(&c, &["lrange", "mylist", "0", "-1"]).await
  970. );
  971. }
  972. #[tokio::test]
  973. async fn brpop_no_waiting() {
  974. let (mut recv, c) = create_connection_and_pubsub();
  975. assert_eq!(
  976. Ok(Value::Integer(5)),
  977. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await,
  978. );
  979. assert_eq!(
  980. Ok(Value::Ignore),
  981. run_command(&c, &["brpop", "foo", "1"]).await
  982. );
  983. assert_eq!(
  984. Some(Value::Array(vec![
  985. Value::Blob("foo".into()),
  986. Value::Blob("5".into()),
  987. ])),
  988. recv.recv().await,
  989. );
  990. }
  991. #[tokio::test]
  992. async fn brpop_timeout() {
  993. let (mut recv, c) = create_connection_and_pubsub();
  994. let x = Instant::now();
  995. assert_eq!(
  996. Ok(Value::Ignore),
  997. run_command(&c, &["brpop", "foobar", "1"]).await
  998. );
  999. assert_eq!(Some(Value::Null), recv.recv().await,);
  1000. assert!(Instant::now() - x >= Duration::from_millis(1000));
  1001. }
  1002. #[tokio::test]
  1003. async fn brpop_wait_insert() {
  1004. let (mut recv, c) = create_connection_and_pubsub();
  1005. let x = Instant::now();
  1006. // Query command that will block connection until some data is inserted
  1007. // to foobar, foo, bar or the 5 seconds timeout happens.
  1008. //
  1009. // We are issuing the command, sleeping a little bit then adding data to
  1010. // foobar, before actually waiting on the result.
  1011. assert_eq!(
  1012. Ok(Value::Ignore),
  1013. run_command(&c, &["brpop", "foobar", "foo", "bar", "5"]).await
  1014. );
  1015. // Sleep 1 second before inserting new data
  1016. sleep(Duration::from_millis(1000)).await;
  1017. assert_eq!(
  1018. Ok(Value::Integer(5)),
  1019. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await,
  1020. );
  1021. // Read the output of the first blpop command now.
  1022. assert_eq!(
  1023. Some(Value::Array(vec![
  1024. Value::Blob("foo".into()),
  1025. Value::Blob("5".into()),
  1026. ])),
  1027. recv.recv().await,
  1028. );
  1029. assert!(Instant::now() - x > Duration::from_millis(1000));
  1030. assert!(Instant::now() - x < Duration::from_millis(5000));
  1031. }
  1032. #[tokio::test]
  1033. async fn lindex() {
  1034. let c = create_connection();
  1035. assert_eq!(
  1036. Ok(Value::Integer(5)),
  1037. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await
  1038. );
  1039. assert_eq!(
  1040. Ok(Value::Array(vec![
  1041. Value::Blob("5".into()),
  1042. Value::Blob("4".into()),
  1043. Value::Blob("3".into()),
  1044. Value::Blob("2".into()),
  1045. Value::Blob("1".into()),
  1046. ])),
  1047. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1048. );
  1049. assert_eq!(
  1050. Ok(Value::Blob("5".into())),
  1051. run_command(&c, &["lindex", "foo", "0"]).await
  1052. );
  1053. assert_eq!(
  1054. Ok(Value::Blob("1".into())),
  1055. run_command(&c, &["lindex", "foo", "-1"]).await
  1056. );
  1057. assert_eq!(
  1058. Ok(Value::Null),
  1059. run_command(&c, &["lindex", "foo", "-100"]).await
  1060. );
  1061. assert_eq!(
  1062. Ok(Value::Null),
  1063. run_command(&c, &["lindex", "foo", "100"]).await
  1064. );
  1065. }
  1066. #[tokio::test]
  1067. async fn linsert_syntax_err() {
  1068. let c = create_connection();
  1069. assert_eq!(
  1070. Ok(Value::Integer(2)),
  1071. run_command(&c, &["rpush", "foo", "hello", "world"]).await
  1072. );
  1073. assert_eq!(
  1074. Err(Error::Syntax),
  1075. run_command(&c, &["linsert", "foo", "beforex", "world", "there"]).await
  1076. );
  1077. }
  1078. #[tokio::test]
  1079. async fn linsert_before() {
  1080. let c = create_connection();
  1081. assert_eq!(
  1082. Ok(Value::Integer(2)),
  1083. run_command(&c, &["rpush", "foo", "hello", "world"]).await
  1084. );
  1085. assert_eq!(
  1086. Ok(Value::Integer(3)),
  1087. run_command(&c, &["linsert", "foo", "before", "world", "there"]).await
  1088. );
  1089. assert_eq!(
  1090. Ok(Value::Array(vec![
  1091. Value::Blob("hello".into()),
  1092. Value::Blob("there".into()),
  1093. Value::Blob("world".into()),
  1094. ])),
  1095. run_command(&c, &["lrange", "foo", "0", "-1"]).await,
  1096. );
  1097. }
  1098. #[tokio::test]
  1099. async fn linsert_after() {
  1100. let c = create_connection();
  1101. assert_eq!(
  1102. Ok(Value::Integer(2)),
  1103. run_command(&c, &["rpush", "foo", "hello", "world"]).await
  1104. );
  1105. assert_eq!(
  1106. Ok(Value::Integer(3)),
  1107. run_command(&c, &["linsert", "foo", "after", "world", "there"]).await
  1108. );
  1109. assert_eq!(
  1110. Ok(Value::Array(vec![
  1111. Value::Blob("hello".into()),
  1112. Value::Blob("world".into()),
  1113. Value::Blob("there".into()),
  1114. ])),
  1115. run_command(&c, &["lrange", "foo", "0", "-1"]).await,
  1116. );
  1117. }
  1118. #[tokio::test]
  1119. async fn linsert_before_after() {
  1120. let c = create_connection();
  1121. assert_eq!(
  1122. Ok(Value::Integer(2)),
  1123. run_command(&c, &["rpush", "foo", "hello", "world"]).await
  1124. );
  1125. assert_eq!(
  1126. Ok(Value::Integer(3)),
  1127. run_command(&c, &["linsert", "foo", "after", "world", "there1"]).await
  1128. );
  1129. assert_eq!(
  1130. Ok(Value::Integer(4)),
  1131. run_command(&c, &["linsert", "foo", "before", "world", "there2"]).await
  1132. );
  1133. assert_eq!(
  1134. Ok(Value::Array(vec![
  1135. Value::Blob("hello".into()),
  1136. Value::Blob("there2".into()),
  1137. Value::Blob("world".into()),
  1138. Value::Blob("there1".into()),
  1139. ])),
  1140. run_command(&c, &["lrange", "foo", "0", "-1"]).await,
  1141. );
  1142. }
  1143. #[tokio::test]
  1144. async fn linsert_not_found() {
  1145. let c = create_connection();
  1146. assert_eq!(
  1147. Ok(Value::Integer(2)),
  1148. run_command(&c, &["rpush", "foo", "hello", "world"]).await
  1149. );
  1150. assert_eq!(
  1151. Ok(Value::Integer(-1)),
  1152. run_command(&c, &["linsert", "foo", "after", "worldx", "there"]).await
  1153. );
  1154. assert_eq!(
  1155. Ok(Value::Integer(-1)),
  1156. run_command(&c, &["linsert", "foo", "before", "worldx", "there"]).await
  1157. );
  1158. }
  1159. #[tokio::test]
  1160. async fn llen() {
  1161. let c = create_connection();
  1162. assert_eq!(
  1163. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await,
  1164. run_command(&c, &["llen", "foo"]).await
  1165. );
  1166. assert_eq!(
  1167. Ok(Value::Integer(0)),
  1168. run_command(&c, &["llen", "foobar"]).await
  1169. );
  1170. }
  1171. #[tokio::test]
  1172. async fn lmove_1() {
  1173. let c = create_connection();
  1174. assert_eq!(
  1175. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await,
  1176. run_command(&c, &["llen", "foo"]).await
  1177. );
  1178. assert_eq!(
  1179. Ok(Value::Blob("1".into())),
  1180. run_command(&c, &["lmove", "foo", "bar", "left", "left"]).await
  1181. );
  1182. assert_eq!(
  1183. Ok(Value::Array(vec![Value::Blob("1".into()),])),
  1184. run_command(&c, &["lrange", "bar", "0", "-1"]).await
  1185. );
  1186. assert_eq!(
  1187. Ok(Value::Blob("5".into())),
  1188. run_command(&c, &["lmove", "foo", "bar", "right", "left"]).await
  1189. );
  1190. assert_eq!(
  1191. Ok(Value::Array(vec![
  1192. Value::Blob("5".into()),
  1193. Value::Blob("1".into()),
  1194. ])),
  1195. run_command(&c, &["lrange", "bar", "0", "-1"]).await
  1196. );
  1197. }
  1198. #[tokio::test]
  1199. async fn lpop() {
  1200. let c = create_connection();
  1201. assert_eq!(
  1202. Ok(Value::Integer(5)),
  1203. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await,
  1204. );
  1205. assert_eq!(
  1206. Ok(Value::Blob("5".into())),
  1207. run_command(&c, &["lpop", "foo"]).await
  1208. );
  1209. assert_eq!(
  1210. Ok(Value::Array(vec![Value::Blob("4".into())])),
  1211. run_command(&c, &["lpop", "foo", "1"]).await
  1212. );
  1213. assert_eq!(
  1214. Ok(Value::Array(vec![
  1215. Value::Blob("3".into()),
  1216. Value::Blob("2".into()),
  1217. Value::Blob("1".into()),
  1218. ])),
  1219. run_command(&c, &["lpop", "foo", "55"]).await
  1220. );
  1221. assert_eq!(
  1222. Ok(Value::Null),
  1223. run_command(&c, &["lpop", "foo", "55"]).await
  1224. );
  1225. assert_eq!(Ok(Value::Null), run_command(&c, &["lpop", "foo"]).await);
  1226. assert_eq!(
  1227. Ok(Value::Integer(0)),
  1228. run_command(&c, &["llen", "foobar"]).await
  1229. );
  1230. }
  1231. #[tokio::test]
  1232. async fn lpos_single_match() {
  1233. let c = create_connection();
  1234. assert_eq!(
  1235. Ok(Value::Integer(11)),
  1236. run_command(
  1237. &c,
  1238. &["RPUSH", "mylist", "a", "b", "c", "d", "1", "2", "3", "4", "3", "3", "3"]
  1239. )
  1240. .await
  1241. );
  1242. assert_eq!(
  1243. Ok(Value::Integer(6)),
  1244. run_command(&c, &["lpos", "mylist", "3"]).await
  1245. );
  1246. }
  1247. #[tokio::test]
  1248. async fn lpos_with_negative_rank_with_count() {
  1249. let c = create_connection();
  1250. assert_eq!(
  1251. Ok(Value::Integer(8)),
  1252. run_command(
  1253. &c,
  1254. &["RPUSH", "mylist", "a", "b", "c", "1", "2", "3", "c", "c"]
  1255. )
  1256. .await
  1257. );
  1258. assert_eq!(
  1259. Ok(Value::Array(vec![Value::Integer(7), Value::Integer(6)])),
  1260. run_command(&c, &["lpos", "mylist", "c", "count", "2", "rank", "-1"]).await
  1261. );
  1262. }
  1263. #[tokio::test]
  1264. async fn lpos_with_negative_rank_with_count_max_len() {
  1265. let c = create_connection();
  1266. assert_eq!(
  1267. Ok(Value::Integer(8)),
  1268. run_command(
  1269. &c,
  1270. &["RPUSH", "mylist", "a", "b", "c", "1", "2", "3", "c", "c"]
  1271. )
  1272. .await
  1273. );
  1274. assert_eq!(
  1275. Ok(Value::Array(vec![Value::Integer(7), Value::Integer(6)])),
  1276. run_command(
  1277. &c,
  1278. &["lpos", "mylist", "c", "count", "0", "maxlen", "3", "rank", "-1"]
  1279. )
  1280. .await
  1281. );
  1282. }
  1283. #[tokio::test]
  1284. async fn lpos_rank_with_count() {
  1285. let c = create_connection();
  1286. assert_eq!(
  1287. Ok(Value::Integer(8)),
  1288. run_command(
  1289. &c,
  1290. &["RPUSH", "mylist", "a", "b", "c", "1", "2", "3", "c", "c"]
  1291. )
  1292. .await
  1293. );
  1294. assert_eq!(
  1295. Ok(Value::Array(vec![Value::Integer(6), Value::Integer(7)])),
  1296. run_command(&c, &["lpos", "mylist", "c", "count", "0", "rank", "2"]).await
  1297. );
  1298. }
  1299. #[tokio::test]
  1300. async fn lpos_all_settings() {
  1301. let c = create_connection();
  1302. assert_eq!(
  1303. Ok(Value::Integer(8)),
  1304. run_command(
  1305. &c,
  1306. &["RPUSH", "mylist", "a", "b", "c", "1", "2", "3", "c", "c"]
  1307. )
  1308. .await
  1309. );
  1310. assert_eq!(
  1311. Ok(Value::Array(vec![Value::Integer(6)])),
  1312. run_command(
  1313. &c,
  1314. &["lpos", "mylist", "c", "count", "0", "rank", "2", "maxlen", "7"]
  1315. )
  1316. .await
  1317. );
  1318. }
  1319. #[tokio::test]
  1320. async fn lpos_negative_rank() {
  1321. let c = create_connection();
  1322. assert_eq!(
  1323. Ok(Value::Integer(8)),
  1324. run_command(
  1325. &c,
  1326. &["RPUSH", "mylist", "a", "b", "c", "1", "2", "3", "c", "c"]
  1327. )
  1328. .await
  1329. );
  1330. assert_eq!(
  1331. Ok(Value::Integer(7)),
  1332. run_command(&c, &["lpos", "mylist", "c", "rank", "-1"]).await
  1333. );
  1334. }
  1335. #[tokio::test]
  1336. async fn lpos_single_skip() {
  1337. let c = create_connection();
  1338. assert_eq!(
  1339. Ok(Value::Integer(11)),
  1340. run_command(
  1341. &c,
  1342. &["RPUSH", "mylist", "a", "b", "c", "d", "1", "2", "3", "4", "3", "3", "3"]
  1343. )
  1344. .await
  1345. );
  1346. assert_eq!(
  1347. Ok(Value::Integer(8)),
  1348. run_command(&c, &["lpos", "mylist", "3", "rank", "2"]).await
  1349. );
  1350. }
  1351. #[tokio::test]
  1352. async fn lpos_single_skip_max_len() {
  1353. let c = create_connection();
  1354. assert_eq!(
  1355. Ok(Value::Integer(11)),
  1356. run_command(
  1357. &c,
  1358. &["RPUSH", "mylist", "a", "b", "c", "d", "1", "2", "3", "4", "3", "3", "3"]
  1359. )
  1360. .await
  1361. );
  1362. assert_eq!(
  1363. Ok(Value::Null),
  1364. run_command(&c, &["lpos", "mylist", "3", "rank", "2", "maxlen", "7"]).await
  1365. );
  1366. }
  1367. #[tokio::test]
  1368. async fn lpos_not_found() {
  1369. let c = create_connection();
  1370. assert_eq!(
  1371. Ok(Value::Array(vec![])),
  1372. run_command(&c, &["lpos", "mylist", "3", "count", "5", "maxlen", "9"]).await
  1373. );
  1374. assert_eq!(
  1375. Ok(Value::Null),
  1376. run_command(&c, &["lpos", "mylist", "3"]).await
  1377. );
  1378. }
  1379. #[tokio::test]
  1380. async fn lpos() {
  1381. let c = create_connection();
  1382. assert_eq!(
  1383. Ok(Value::Integer(11)),
  1384. run_command(
  1385. &c,
  1386. &["RPUSH", "mylist", "a", "b", "c", "d", "1", "2", "3", "4", "3", "3", "3"]
  1387. )
  1388. .await
  1389. );
  1390. assert_eq!(
  1391. Ok(Value::Array(vec![Value::Integer(6), Value::Integer(8),])),
  1392. run_command(&c, &["lpos", "mylist", "3", "count", "5", "maxlen", "9"]).await
  1393. );
  1394. }
  1395. #[tokio::test]
  1396. async fn lpush() {
  1397. let c = create_connection();
  1398. assert_eq!(
  1399. Ok(Value::Integer(5)),
  1400. run_command(&c, &["lpush", "foo", "1", "2", "3", "4", "5"]).await
  1401. );
  1402. assert_eq!(
  1403. Ok(Value::Array(vec![
  1404. Value::Blob("5".into()),
  1405. Value::Blob("4".into()),
  1406. Value::Blob("3".into()),
  1407. Value::Blob("2".into()),
  1408. Value::Blob("1".into()),
  1409. ])),
  1410. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1411. );
  1412. assert_eq!(
  1413. Ok(Value::Integer(10)),
  1414. run_command(&c, &["lpush", "foo", "6", "7", "8", "9", "10"]).await
  1415. );
  1416. assert_eq!(
  1417. Ok(Value::Array(vec![
  1418. Value::Blob("10".into()),
  1419. Value::Blob("9".into()),
  1420. Value::Blob("8".into()),
  1421. Value::Blob("7".into()),
  1422. Value::Blob("6".into()),
  1423. Value::Blob("5".into()),
  1424. Value::Blob("4".into()),
  1425. Value::Blob("3".into()),
  1426. Value::Blob("2".into()),
  1427. Value::Blob("1".into()),
  1428. ])),
  1429. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1430. );
  1431. }
  1432. #[tokio::test]
  1433. async fn lpush_simple() {
  1434. let c = create_connection();
  1435. assert_eq!(
  1436. Ok(Value::Integer(1)),
  1437. run_command(&c, &["lpush", "foo", "world"]).await
  1438. );
  1439. assert_eq!(
  1440. Ok(Value::Integer(2)),
  1441. run_command(&c, &["lpush", "foo", "hello"]).await
  1442. );
  1443. assert_eq!(
  1444. Ok(Value::Array(vec![
  1445. Value::Blob("hello".into()),
  1446. Value::Blob("world".into()),
  1447. ])),
  1448. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1449. );
  1450. }
  1451. #[tokio::test]
  1452. async fn lset() {
  1453. let c = create_connection();
  1454. assert_eq!(
  1455. Ok(Value::Integer(5)),
  1456. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1457. );
  1458. assert_eq!(
  1459. Ok(Value::Ok),
  1460. run_command(&c, &["lset", "foo", "-1", "6"]).await,
  1461. );
  1462. assert_eq!(
  1463. Ok(Value::Ok),
  1464. run_command(&c, &["lset", "foo", "-2", "7"]).await,
  1465. );
  1466. assert_eq!(
  1467. Ok(Value::Ok),
  1468. run_command(&c, &["lset", "foo", "0", "8"]).await,
  1469. );
  1470. assert_eq!(
  1471. Err(Error::OutOfRange),
  1472. run_command(&c, &["lset", "foo", "55", "8"]).await,
  1473. );
  1474. assert_eq!(
  1475. Err(Error::OutOfRange),
  1476. run_command(&c, &["lset", "foo", "-55", "8"]).await,
  1477. );
  1478. assert_eq!(
  1479. Err(Error::NotFound),
  1480. run_command(&c, &["lset", "key_not_exists", "-55", "8"]).await,
  1481. );
  1482. assert_eq!(
  1483. Ok(Value::Blob("6".into())),
  1484. run_command(&c, &["rpop", "foo"]).await
  1485. );
  1486. assert_eq!(
  1487. Ok(Value::Blob("7".into())),
  1488. run_command(&c, &["rpop", "foo"]).await
  1489. );
  1490. assert_eq!(
  1491. Ok(Value::Blob("8".into())),
  1492. run_command(&c, &["lpop", "foo"]).await
  1493. );
  1494. }
  1495. #[tokio::test]
  1496. async fn ltrim() {
  1497. let c = create_connection();
  1498. assert_eq!(
  1499. Ok(Value::Integer(5)),
  1500. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1501. );
  1502. assert_eq!(
  1503. Ok(Value::Ok),
  1504. run_command(&c, &["ltrim", "foo", "1", "-2"]).await
  1505. );
  1506. assert_eq!(
  1507. Ok(Value::Array(vec![
  1508. Value::Blob("2".into()),
  1509. Value::Blob("3".into()),
  1510. Value::Blob("4".into()),
  1511. ])),
  1512. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1513. );
  1514. }
  1515. #[tokio::test]
  1516. async fn rpop() {
  1517. let c = create_connection();
  1518. assert_eq!(
  1519. Ok(Value::Integer(5)),
  1520. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1521. );
  1522. assert_eq!(
  1523. Ok(Value::Blob("5".into())),
  1524. run_command(&c, &["rpop", "foo"]).await
  1525. );
  1526. assert_eq!(
  1527. Ok(Value::Array(vec![Value::Blob("4".into())])),
  1528. run_command(&c, &["rpop", "foo", "1"]).await
  1529. );
  1530. assert_eq!(
  1531. Ok(Value::Array(vec![
  1532. Value::Blob("3".into()),
  1533. Value::Blob("2".into()),
  1534. Value::Blob("1".into()),
  1535. ])),
  1536. run_command(&c, &["rpop", "foo", "55"]).await
  1537. );
  1538. assert_eq!(
  1539. Ok(Value::Null),
  1540. run_command(&c, &["rpop", "foo", "55"]).await
  1541. );
  1542. assert_eq!(Ok(Value::Null), run_command(&c, &["rpop", "foo"]).await);
  1543. assert_eq!(
  1544. Ok(Value::Integer(0)),
  1545. run_command(&c, &["llen", "foobar"]).await
  1546. );
  1547. }
  1548. #[tokio::test]
  1549. async fn rpush_simple() {
  1550. let c = create_connection();
  1551. assert_eq!(
  1552. Ok(Value::Integer(1)),
  1553. run_command(&c, &["rpush", "foo", "world"]).await
  1554. );
  1555. assert_eq!(
  1556. Ok(Value::Integer(2)),
  1557. run_command(&c, &["rpush", "foo", "hello"]).await
  1558. );
  1559. assert_eq!(
  1560. Ok(Value::Array(vec![
  1561. Value::Blob("world".into()),
  1562. Value::Blob("hello".into()),
  1563. ])),
  1564. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1565. );
  1566. }
  1567. #[tokio::test]
  1568. async fn lrange() {
  1569. let c = create_connection();
  1570. assert_eq!(
  1571. Ok(Value::Integer(5)),
  1572. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1573. );
  1574. assert_eq!(
  1575. Ok(Value::Array(vec![
  1576. Value::Blob("1".into()),
  1577. Value::Blob("2".into()),
  1578. Value::Blob("3".into()),
  1579. Value::Blob("4".into()),
  1580. Value::Blob("5".into()),
  1581. ])),
  1582. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1583. );
  1584. assert_eq!(
  1585. Ok(Value::Array(vec![
  1586. Value::Blob("1".into()),
  1587. Value::Blob("2".into()),
  1588. Value::Blob("3".into()),
  1589. Value::Blob("4".into()),
  1590. ])),
  1591. run_command(&c, &["lrange", "foo", "0", "-2"]).await
  1592. );
  1593. assert_eq!(
  1594. Ok(Value::Array(vec![
  1595. Value::Blob("4".into()),
  1596. Value::Blob("5".into()),
  1597. ])),
  1598. run_command(&c, &["lrange", "foo", "-2", "-1"]).await
  1599. );
  1600. assert_eq!(
  1601. Ok(Value::Array(vec![Value::Blob("3".into()),])),
  1602. run_command(&c, &["lrange", "foo", "-3", "-3"]).await
  1603. );
  1604. }
  1605. #[tokio::test]
  1606. async fn rpush() {
  1607. let c = create_connection();
  1608. assert_eq!(
  1609. Ok(Value::Integer(5)),
  1610. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1611. );
  1612. assert_eq!(
  1613. Ok(Value::Array(vec![
  1614. Value::Blob("1".into()),
  1615. Value::Blob("2".into()),
  1616. Value::Blob("3".into()),
  1617. Value::Blob("4".into()),
  1618. Value::Blob("5".into()),
  1619. ])),
  1620. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1621. );
  1622. assert_eq!(
  1623. Ok(Value::Integer(10)),
  1624. run_command(&c, &["rpush", "foo", "6", "7", "8", "9", "10"]).await
  1625. );
  1626. assert_eq!(
  1627. Ok(Value::Array(vec![
  1628. Value::Blob("1".into()),
  1629. Value::Blob("2".into()),
  1630. Value::Blob("3".into()),
  1631. Value::Blob("4".into()),
  1632. Value::Blob("5".into()),
  1633. Value::Blob("6".into()),
  1634. Value::Blob("7".into()),
  1635. Value::Blob("8".into()),
  1636. Value::Blob("9".into()),
  1637. Value::Blob("10".into()),
  1638. ])),
  1639. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1640. );
  1641. }
  1642. #[tokio::test]
  1643. async fn rpushx() {
  1644. let c = create_connection();
  1645. assert_eq!(
  1646. Ok(Value::Integer(5)),
  1647. run_command(&c, &["rpush", "foo", "1", "2", "3", "4", "5"]).await
  1648. );
  1649. assert_eq!(
  1650. Ok(Value::Array(vec![
  1651. Value::Blob("1".into()),
  1652. Value::Blob("2".into()),
  1653. Value::Blob("3".into()),
  1654. Value::Blob("4".into()),
  1655. Value::Blob("5".into()),
  1656. ])),
  1657. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1658. );
  1659. assert_eq!(
  1660. Ok(Value::Integer(10)),
  1661. run_command(&c, &["rpushx", "foo", "6", "7", "8", "9", "10"]).await
  1662. );
  1663. assert_eq!(
  1664. Ok(Value::Array(vec![
  1665. Value::Blob("1".into()),
  1666. Value::Blob("2".into()),
  1667. Value::Blob("3".into()),
  1668. Value::Blob("4".into()),
  1669. Value::Blob("5".into()),
  1670. Value::Blob("6".into()),
  1671. Value::Blob("7".into()),
  1672. Value::Blob("8".into()),
  1673. Value::Blob("9".into()),
  1674. Value::Blob("10".into()),
  1675. ])),
  1676. run_command(&c, &["lrange", "foo", "0", "-1"]).await
  1677. );
  1678. assert_eq!(
  1679. Ok(Value::Integer(0)),
  1680. run_command(&c, &["rpushx", "foobar", "6", "7", "8", "9", "10"]).await
  1681. );
  1682. }
  1683. #[tokio::test]
  1684. async fn lrange_test_1() {
  1685. let c = create_connection();
  1686. assert_eq!(
  1687. Ok(Value::Integer(10)),
  1688. run_command(
  1689. &c,
  1690. &[
  1691. "rpush",
  1692. "mylist",
  1693. "largevalue",
  1694. "1",
  1695. "2",
  1696. "3",
  1697. "4",
  1698. "5",
  1699. "6",
  1700. "7",
  1701. "8",
  1702. "9"
  1703. ]
  1704. )
  1705. .await
  1706. );
  1707. assert_eq!(
  1708. Ok(Value::Array(vec![
  1709. "1".into(),
  1710. "2".into(),
  1711. "3".into(),
  1712. "4".into(),
  1713. "5".into(),
  1714. "6".into(),
  1715. "7".into(),
  1716. "8".into()
  1717. ])),
  1718. run_command(&c, &["lrange", "mylist", "1", "-2"]).await
  1719. );
  1720. assert_eq!(
  1721. Ok(Value::Array(vec!["7".into(), "8".into(), "9".into()])),
  1722. run_command(&c, &["lrange", "mylist", "-3", "-1"]).await
  1723. );
  1724. assert_eq!(
  1725. Ok(Value::Array(vec!["4".into()])),
  1726. run_command(&c, &["lrange", "mylist", "4", "4"]).await
  1727. );
  1728. }
  1729. }