list.rs 55 KB

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