@Test public void shouldCleanupSegmentsThatHaveExpired() throws Exception { final Segment segment1 = segments.getOrCreateSegment(0, context); final Segment segment2 = segments.getOrCreateSegment(0, context); final Segment segment3 = segments.getOrCreateSegment(7, context); assertFalse(segment1.isOpen()); assertFalse(segment2.isOpen()); assertTrue(segment3.isOpen()); assertFalse(new File(context.stateDir(), "test/test-197001010000").exists()); assertFalse(new File(context.stateDir(), "test/test-197001010001").exists()); assertTrue(new File(context.stateDir(), "test/test-197001010007").exists()); }
Segment getOrCreateSegment(final long segmentId, final ProcessorContext context) { if (segmentId > maxSegmentId - numSegments) { final long key = segmentId % numSegments; final Segment segment = segments.get(key); if (!isSegment(segment, segmentId)) { cleanup(segmentId); } Segment newSegment = new Segment(segmentName(segmentId), name, segmentId); Segment previousSegment = segments.putIfAbsent(key, newSegment); if (previousSegment == null) { newSegment.openDB(context); maxSegmentId = segmentId > maxSegmentId ? segmentId : maxSegmentId; if (minSegmentId == Long.MAX_VALUE) { minSegmentId = maxSegmentId; } } return previousSegment == null ? newSegment : previousSegment; } else { return null; } }
Segments { Segment getOrCreateSegment(final long segmentId, final ProcessorContext context) { if (segmentId > maxSegmentId - numSegments) { final long key = segmentId % numSegments; final Segment segment = segments.get(key); if (!isSegment(segment, segmentId)) { cleanup(segmentId); } Segment newSegment = new Segment(segmentName(segmentId), name, segmentId); Segment previousSegment = segments.putIfAbsent(key, newSegment); if (previousSegment == null) { newSegment.openDB(context); maxSegmentId = segmentId > maxSegmentId ? segmentId : maxSegmentId; if (minSegmentId == Long.MAX_VALUE) { minSegmentId = maxSegmentId; } } return previousSegment == null ? newSegment : previousSegment; } else { return null; } } }
Segments { Segment getOrCreateSegment(final long segmentId, final ProcessorContext context) { if (segmentId > maxSegmentId - numSegments) { final long key = segmentId % numSegments; final Segment segment = segments.get(key); if (!isSegment(segment, segmentId)) { cleanup(segmentId); } Segment newSegment = new Segment(segmentName(segmentId), name, segmentId); Segment previousSegment = segments.putIfAbsent(key, newSegment); if (previousSegment == null) { newSegment.openDB(context); maxSegmentId = segmentId > maxSegmentId ? segmentId : maxSegmentId; if (minSegmentId == Long.MAX_VALUE) { minSegmentId = maxSegmentId; } } return previousSegment == null ? newSegment : previousSegment; } else { return null; } } Segments(final String name, final long retentionPeriod, final int numSegments); }
Segments { Segment getOrCreateSegment(final long segmentId, final ProcessorContext context) { if (segmentId > maxSegmentId - numSegments) { final long key = segmentId % numSegments; final Segment segment = segments.get(key); if (!isSegment(segment, segmentId)) { cleanup(segmentId); } Segment newSegment = new Segment(segmentName(segmentId), name, segmentId); Segment previousSegment = segments.putIfAbsent(key, newSegment); if (previousSegment == null) { newSegment.openDB(context); maxSegmentId = segmentId > maxSegmentId ? segmentId : maxSegmentId; if (minSegmentId == Long.MAX_VALUE) { minSegmentId = maxSegmentId; } } return previousSegment == null ? newSegment : previousSegment; } else { return null; } } Segments(final String name, final long retentionPeriod, final int numSegments); void close(); }
Segments { Segment getOrCreateSegment(final long segmentId, final ProcessorContext context) { if (segmentId > maxSegmentId - numSegments) { final long key = segmentId % numSegments; final Segment segment = segments.get(key); if (!isSegment(segment, segmentId)) { cleanup(segmentId); } Segment newSegment = new Segment(segmentName(segmentId), name, segmentId); Segment previousSegment = segments.putIfAbsent(key, newSegment); if (previousSegment == null) { newSegment.openDB(context); maxSegmentId = segmentId > maxSegmentId ? segmentId : maxSegmentId; if (minSegmentId == Long.MAX_VALUE) { minSegmentId = maxSegmentId; } } return previousSegment == null ? newSegment : previousSegment; } else { return null; } } Segments(final String name, final long retentionPeriod, final int numSegments); void close();  }