### Question:
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();  }### Answer:
@Test public void shouldCreateSegments() throws Exception { final Segment segment1 = segments.getOrCreateSegment(0, context); final Segment segment2 = segments.getOrCreateSegment(1, context); final Segment segment3 = segments.getOrCreateSegment(2, context); assertTrue(new File(context.stateDir(), "test/test-197001010000").isDirectory()); assertTrue(new File(context.stateDir(), "test/test-197001010001").isDirectory()); assertTrue(new File(context.stateDir(), "test/test-197001010002").isDirectory()); assertEquals(true, segment1.isOpen()); assertEquals(true, segment2.isOpen()); assertEquals(true, segment3.isOpen()); }
@Test public void shouldNotCreateSegmentThatIsAlreadyExpired() throws Exception { segments.getOrCreateSegment(7, context); assertNull(segments.getOrCreateSegment(0, context)); assertFalse(new File(context.stateDir(), "test/test-197001010000").exists()); }
@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()); }
@Test public void shouldRollSegments() throws Exception { segments.getOrCreateSegment(0, context); verifyCorrectSegments(0, 1); segments.getOrCreateSegment(1, context); verifyCorrectSegments(0, 2); segments.getOrCreateSegment(2, context); verifyCorrectSegments(0, 3); segments.getOrCreateSegment(3, context); verifyCorrectSegments(0, 4); segments.getOrCreateSegment(4, context); verifyCorrectSegments(0, 5); segments.getOrCreateSegment(5, context); verifyCorrectSegments(1, 5); segments.getOrCreateSegment(6, context); verifyCorrectSegments(2, 5); }