1 public List<Event> readEvents(int numEvents) throws IOException {
2 if (!committed) {
3 if (!currentFile.isPresent()) {
4 throw new IllegalStateException("File should not roll when " +
5 "commit is outstanding.");
6 }
7 logger.info("Last read was never committed - resetting mark position.");
8 currentFile.get().getDeserializer().reset();
9 } else {
10 // Check if new files have arrived since last call
11 if (!currentFile.isPresent()) {
12 currentFile = getNextFile();
13 }
14 // Return empty list if no new files
15 if (!currentFile.isPresent()) {
16 return Collections.emptyList();
17 }
18 }
19
20 EventDeserializer des = currentFile.get().getDeserializer();
21 List<Event> events = des.readEvents(numEvents);
22
23 /* It's possible that the last read took us just up to a file boundary.
24 * If so, try to roll to the next file, if there is one. */
25 if (events.isEmpty()) {
26 retireCurrentFile();
27 currentFile = getNextFile();
28 if (!currentFile.isPresent()) {
29 return Collections.emptyList();
30 }
31 events = currentFile.get().getDeserializer().readEvents(numEvents);
32 }
33
34 //写入header值,略47
48 committed = false;
49 lastFileRead = currentFile;
50 return events;
51 }
在这个方法中,我们看到了
currentFile:该对象采用了谷歌的Optional进行封装,更加容易判断空指针等等。Optional<FileInfo>,该FileInfo封装了普通的File对象和针对该file对象的EventDeserializer(事件序列器)
该currentFile主要在ReliableSpoolingFileEventReader 类中的Optional<FileInfo> openFile(File file),Optional<FileInfo> getNextFile() 方法中调用。