@Test public void blpop() throws Exception { String head = "blpop0"; RList list = db.getList(); int num = 10000; try { for (int k = 0; k < 10; k++) { for (int i = 0; i < num; i++) { list.add(head, ("hello" + i).getBytes()); } JoinFuture<List> joinFuture = JoinFuture.build(executorService, List.class); for (int i = 0; i < 4; i++) { joinFuture.add(args -> { List<byte[]> listJoin = new ArrayList<>(); try { while (true) { List<byte[]> pops = list.blpop(head, 1000); Thread.sleep(50); if (pops.size() == 0) { break; } listJoin.addAll(pops); } } catch (Exception e) { e.printStackTrace(); } return listJoin; }); } List<List> joins = joinFuture.join(); List<String> all = new ArrayList<>(); for (List<byte[]> join : joins) { for (byte[] bytes : join) { all.add(new String(bytes)); } } Assert.assertEquals(num, all.size()); Set<String> sets = new HashSet<>(all); for (int i = 0; i < num; i++) { Assert.assertTrue(sets.contains("hello" + i)); } } } finally { list.delete(head); } }
public List<byte[]> blpop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { byte[] key_b = getKey(key); List<byte[]> list = new ArrayList<>(); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return list; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { final int maxCount = num > 0 ? num : Integer.MAX_VALUE; ValueK valueK_seek = new ValueK(key_b.length, key_b, metaV.getVersion(), metaV.left); List<byte[]> delete_keys = new ArrayList<>(); ValueKD valueKD = valueK_seek.convertValueBytes(); byte[] heads = valueKD.toHeadBytes(); iterator.seek(valueKD.toBytes()); int count = 0; while (iterator.isValid() && count++ < maxCount) { byte[] ikey = iterator.key(); if (!BytesUtil.checkHead(heads, ikey)) break; ValueKD key_bytes = ValueKD.build(ikey); delete_keys.add(ikey); list.add(iterator.value()); metaV.setLeft(key_bytes.getIndexV()); iterator.next(); } metaV.setSize(metaV.getSize() - delete_keys.size()); if (metaV.getSize() != 0) { byte[] ikey = iterator.key(); ValueKD key_bytes = ValueKD.build(ikey); metaV.setLeft(key_bytes.getIndexV()); } start(); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); for (byte[] delete_key : delete_keys) { deleteDB(delete_key, SstColumnFamily.DEFAULT); } commit(); checkTxCommit(); return list; } catch (Exception e) { throw e; } finally { release(); } } finally { unlock(lockEntity); } } catch (Exception e) { checkTxRollBack(); throw e; } }
RList extends RCollection { public List<byte[]> blpop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { byte[] key_b = getKey(key); List<byte[]> list = new ArrayList<>(); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return list; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { final int maxCount = num > 0 ? num : Integer.MAX_VALUE; ValueK valueK_seek = new ValueK(key_b.length, key_b, metaV.getVersion(), metaV.left); List<byte[]> delete_keys = new ArrayList<>(); ValueKD valueKD = valueK_seek.convertValueBytes(); byte[] heads = valueKD.toHeadBytes(); iterator.seek(valueKD.toBytes()); int count = 0; while (iterator.isValid() && count++ < maxCount) { byte[] ikey = iterator.key(); if (!BytesUtil.checkHead(heads, ikey)) break; ValueKD key_bytes = ValueKD.build(ikey); delete_keys.add(ikey); list.add(iterator.value()); metaV.setLeft(key_bytes.getIndexV()); iterator.next(); } metaV.setSize(metaV.getSize() - delete_keys.size()); if (metaV.getSize() != 0) { byte[] ikey = iterator.key(); ValueKD key_bytes = ValueKD.build(ikey); metaV.setLeft(key_bytes.getIndexV()); } start(); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); for (byte[] delete_key : delete_keys) { deleteDB(delete_key, SstColumnFamily.DEFAULT); } commit(); checkTxCommit(); return list; } catch (Exception e) { throw e; } finally { release(); } } finally { unlock(lockEntity); } } catch (Exception e) { checkTxRollBack(); throw e; } } }
RList extends RCollection { public List<byte[]> blpop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { byte[] key_b = getKey(key); List<byte[]> list = new ArrayList<>(); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return list; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { final int maxCount = num > 0 ? num : Integer.MAX_VALUE; ValueK valueK_seek = new ValueK(key_b.length, key_b, metaV.getVersion(), metaV.left); List<byte[]> delete_keys = new ArrayList<>(); ValueKD valueKD = valueK_seek.convertValueBytes(); byte[] heads = valueKD.toHeadBytes(); iterator.seek(valueKD.toBytes()); int count = 0; while (iterator.isValid() && count++ < maxCount) { byte[] ikey = iterator.key(); if (!BytesUtil.checkHead(heads, ikey)) break; ValueKD key_bytes = ValueKD.build(ikey); delete_keys.add(ikey); list.add(iterator.value()); metaV.setLeft(key_bytes.getIndexV()); iterator.next(); } metaV.setSize(metaV.getSize() - delete_keys.size()); if (metaV.getSize() != 0) { byte[] ikey = iterator.key(); ValueKD key_bytes = ValueKD.build(ikey); metaV.setLeft(key_bytes.getIndexV()); } start(); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); for (byte[] delete_key : delete_keys) { deleteDB(delete_key, SstColumnFamily.DEFAULT); } commit(); checkTxCommit(); return list; } catch (Exception e) { throw e; } finally { release(); } } finally { unlock(lockEntity); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RList(DB db); }
RList extends RCollection { public List<byte[]> blpop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { byte[] key_b = getKey(key); List<byte[]> list = new ArrayList<>(); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return list; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { final int maxCount = num > 0 ? num : Integer.MAX_VALUE; ValueK valueK_seek = new ValueK(key_b.length, key_b, metaV.getVersion(), metaV.left); List<byte[]> delete_keys = new ArrayList<>(); ValueKD valueKD = valueK_seek.convertValueBytes(); byte[] heads = valueKD.toHeadBytes(); iterator.seek(valueKD.toBytes()); int count = 0; while (iterator.isValid() && count++ < maxCount) { byte[] ikey = iterator.key(); if (!BytesUtil.checkHead(heads, ikey)) break; ValueKD key_bytes = ValueKD.build(ikey); delete_keys.add(ikey); list.add(iterator.value()); metaV.setLeft(key_bytes.getIndexV()); iterator.next(); } metaV.setSize(metaV.getSize() - delete_keys.size()); if (metaV.getSize() != 0) { byte[] ikey = iterator.key(); ValueKD key_bytes = ValueKD.build(ikey); metaV.setLeft(key_bytes.getIndexV()); } start(); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); for (byte[] delete_key : delete_keys) { deleteDB(delete_key, SstColumnFamily.DEFAULT); } commit(); checkTxCommit(); return list; } catch (Exception e) { throw e; } finally { release(); } } finally { unlock(lockEntity); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RList(DB db); void add(String key, byte[] v); void addAll(String key, List<byte[]> vs); void addAllMayTTL(String key, List<byte[]> vs, int ttl); void set(String key, long i, byte[] v); byte[] get(String key, long i); List<byte[]> get(String key, List<Long> is); Long left(String key); Long right(String key); int size(String key); void deleteFast(String key); void delete(String key); boolean isExist(String key); List<byte[]> range(String key, long start, long end); @Override @SuppressWarnings("unchecked") RIterator<RList> iterator(String key); Entry getEntry(RocksIterator iterator); List<byte[]> blpop(String key, int num); List<byte[]> brpop(String key, int num); @Override KeyIterator getKeyIterator(); void ttl(String key, int ttl); void delTtl(String key); int getTtl(String key); void addMayTTL(String key, byte[] v, int ttl); }
RList extends RCollection { public List<byte[]> blpop(String key, int num) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { LockEntity lockEntity = lock(key); try { byte[] key_b = getKey(key); List<byte[]> list = new ArrayList<>(); MetaV metaV = getMeta(key_b); if (metaV == null) { checkTxCommit(); return list; } try (final RocksIterator iterator = newIterator(SstColumnFamily.DEFAULT)) { final int maxCount = num > 0 ? num : Integer.MAX_VALUE; ValueK valueK_seek = new ValueK(key_b.length, key_b, metaV.getVersion(), metaV.left); List<byte[]> delete_keys = new ArrayList<>(); ValueKD valueKD = valueK_seek.convertValueBytes(); byte[] heads = valueKD.toHeadBytes(); iterator.seek(valueKD.toBytes()); int count = 0; while (iterator.isValid() && count++ < maxCount) { byte[] ikey = iterator.key(); if (!BytesUtil.checkHead(heads, ikey)) break; ValueKD key_bytes = ValueKD.build(ikey); delete_keys.add(ikey); list.add(iterator.value()); metaV.setLeft(key_bytes.getIndexV()); iterator.next(); } metaV.setSize(metaV.getSize() - delete_keys.size()); if (metaV.getSize() != 0) { byte[] ikey = iterator.key(); ValueKD key_bytes = ValueKD.build(ikey); metaV.setLeft(key_bytes.getIndexV()); } start(); putDB(key_b, metaV.convertMetaBytes().toBytes(), SstColumnFamily.META); for (byte[] delete_key : delete_keys) { deleteDB(delete_key, SstColumnFamily.DEFAULT); } commit(); checkTxCommit(); return list; } catch (Exception e) { throw e; } finally { release(); } } finally { unlock(lockEntity); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RList(DB db); void add(String key, byte[] v); void addAll(String key, List<byte[]> vs); void addAllMayTTL(String key, List<byte[]> vs, int ttl); void set(String key, long i, byte[] v); byte[] get(String key, long i); List<byte[]> get(String key, List<Long> is); Long left(String key); Long right(String key); int size(String key); void deleteFast(String key); void delete(String key); boolean isExist(String key); List<byte[]> range(String key, long start, long end); @Override @SuppressWarnings("unchecked") RIterator<RList> iterator(String key); Entry getEntry(RocksIterator iterator); List<byte[]> blpop(String key, int num); List<byte[]> brpop(String key, int num); @Override KeyIterator getKeyIterator(); void ttl(String key, int ttl); void delTtl(String key); int getTtl(String key); void addMayTTL(String key, byte[] v, int ttl); final static String HEAD; final static byte[] HEAD_VALUE_B; final static byte[] HEAD_B; }