PersistentRedisCache extends AbstractPersistentCache { @Override public void writeSegment(long msb, long lsb, Buffer buffer) { String segmentId = new UUID(msb, lsb).toString(); Buffer bufferCopy = buffer.duplicate(); Runnable task = () -> { if (writesPending.add(segmentId)) { final ByteArrayOutputStream bos = new ByteArrayOutputStream(); try (WritableByteChannel channel = Channels.newChannel(bos); Jedis redis = redisPool.getResource()) { while (bufferCopy.hasRemaining()) { bufferCopy.write(channel); } final byte[] key = (REDIS_PREFIX + ":" + segmentId).getBytes(); redis.set(key, bos.toByteArray(), setParamsWithExpire); cacheSize.addAndGet(bos.size()); } catch (Throwable t) { logger.debug("Unable to write segment {} to cache: {}", segmentId, t.getMessage()); } finally { writesPending.remove(segmentId); } } }; executor.execute(task); } PersistentRedisCache(String redisHost, int redisPort, int redisExpireSeconds, int redisSocketTimeout,
            int redisConnectionTimeout, int redisMinConnections, int redisMaxConnections, int redisMaxTotalConnections,
            int redisDBIndex, IOMonitor redisCacheIOMonitor); @Override boolean containsSegment(long msb, long lsb); @Override void writeSegment(long msb, long lsb, Buffer buffer); @Override void cleanUp(); static final int DEFAULT_REDIS_CACHE_EXPIRE_SECONDS; static final String NAME; }
@Test public void testIOMonitor() throws IOException, InterruptedException { UUID segmentUUID = UUID.randomUUID(); long msb = segmentUUID.getMostSignificantBits(); long lsb = segmentUUID.getLeastSignificantBits(); persistentCache.readSegment(msb, lsb, () -> null); verify(ioMonitorAdapter, never()).afterSegmentRead(any(), anyLong(), anyLong(), anyInt(), anyLong()); persistentCache.writeSegment(msb, lsb, Buffer.wrap("segment_content".getBytes())); Thread.sleep(300); persistentCache.readSegment(msb, lsb, () -> null); verify(ioMonitorAdapter, times(1)).afterSegmentRead(any(), eq(msb), eq(lsb), anyInt(), anyLong()); }