43 */ |
43 */ |
44 public final class EventFileStream extends AbstractEventStream { |
44 public final class EventFileStream extends AbstractEventStream { |
45 private final static Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator(); |
45 private final static Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator(); |
46 |
46 |
47 private final RecordingInput input; |
47 private final RecordingInput input; |
48 private ChunkParser chunkParser; |
48 |
49 private RecordedEvent[] sortedList; |
49 private ChunkParser currentParser; |
|
50 private RecordedEvent[] cacheSorted; |
50 |
51 |
51 public EventFileStream(AccessControlContext acc, Path path) throws IOException { |
52 public EventFileStream(AccessControlContext acc, Path path) throws IOException { |
52 super(acc, false); |
53 super(acc, false); |
53 Objects.requireNonNull(path); |
54 Objects.requireNonNull(path); |
54 this.input = new RecordingInput(path.toFile(), FileAccess.UNPRIVILIGED); |
55 this.input = new RecordingInput(path.toFile(), FileAccess.UNPRIVILIGED); |
85 } |
86 } |
86 if (disp.endTime != null) { |
87 if (disp.endTime != null) { |
87 end = disp.endNanos; |
88 end = disp.endNanos; |
88 } |
89 } |
89 |
90 |
90 chunkParser = new ChunkParser(input, disp.parserConfiguration); |
91 currentParser = new ChunkParser(input, disp.parserConfiguration); |
91 while (!isClosed()) { |
92 while (!isClosed()) { |
92 if (chunkParser.getStartNanos() > end) { |
93 if (currentParser.getStartNanos() > end) { |
93 close(); |
94 close(); |
94 return; |
95 return; |
95 } |
96 } |
96 disp = dispatcher(); |
97 disp = dispatcher(); |
97 disp.parserConfiguration.filterStart = start; |
98 disp.parserConfiguration.filterStart = start; |
98 disp.parserConfiguration.filterEnd = end; |
99 disp.parserConfiguration.filterEnd = end; |
99 chunkParser.updateConfiguration(disp.parserConfiguration, true); |
100 currentParser.updateConfiguration(disp.parserConfiguration, true); |
100 chunkParser.setFlushOperation(getFlushOperation()); |
101 currentParser.setFlushOperation(getFlushOperation()); |
101 if (disp.parserConfiguration.ordered) { |
102 if (disp.parserConfiguration.isOrdered()) { |
102 processOrdered(disp); |
103 processOrdered(disp); |
103 } else { |
104 } else { |
104 processUnordered(disp); |
105 processUnordered(disp); |
105 } |
106 } |
106 if (isClosed() || chunkParser.isLastChunk()) { |
107 if (isClosed() || currentParser.isLastChunk()) { |
107 return; |
108 return; |
108 } |
109 } |
109 chunkParser = chunkParser.nextChunkParser(); |
110 currentParser = currentParser.nextChunkParser(); |
110 } |
111 } |
111 } |
112 } |
112 |
113 |
113 private void processOrdered(Dispatcher c) throws IOException { |
114 private void processOrdered(Dispatcher c) throws IOException { |
114 if (sortedList == null) { |
115 if (cacheSorted == null) { |
115 sortedList = new RecordedEvent[10_000]; |
116 cacheSorted = new RecordedEvent[10_000]; |
116 } |
117 } |
117 RecordedEvent event; |
118 RecordedEvent event; |
118 int index = 0; |
119 int index = 0; |
119 while (true) { |
120 while (true) { |
120 event = chunkParser.readEvent(); |
121 event = currentParser.readEvent(); |
121 if (event == null) { |
122 if (event == null) { |
122 Arrays.sort(sortedList, 0, index, EVENT_COMPARATOR); |
123 Arrays.sort(cacheSorted, 0, index, EVENT_COMPARATOR); |
123 for (int i = 0; i < index; i++) { |
124 for (int i = 0; i < index; i++) { |
124 c.dispatch(sortedList[i]); |
125 c.dispatch(cacheSorted[i]); |
125 } |
126 } |
126 return; |
127 return; |
127 } |
128 } |
128 if (index == sortedList.length) { |
129 if (index == cacheSorted.length) { |
129 RecordedEvent[] tmp = sortedList; |
130 RecordedEvent[] tmp = cacheSorted; |
130 sortedList = new RecordedEvent[2 * tmp.length]; |
131 cacheSorted = new RecordedEvent[2 * tmp.length]; |
131 System.arraycopy(tmp, 0, sortedList, 0, tmp.length); |
132 System.arraycopy(tmp, 0, cacheSorted, 0, tmp.length); |
132 } |
133 } |
133 sortedList[index++] = event; |
134 cacheSorted[index++] = event; |
134 } |
135 } |
135 } |
136 } |
136 |
137 |
137 private void processUnordered(Dispatcher c) throws IOException { |
138 private void processUnordered(Dispatcher c) throws IOException { |
138 while (!isClosed()) { |
139 while (!isClosed()) { |
139 RecordedEvent event = chunkParser.readEvent(); |
140 RecordedEvent event = currentParser.readEvent(); |
140 if (event == null) { |
141 if (event == null) { |
141 return; |
142 return; |
142 } |
143 } |
143 c.dispatch(event); |
144 c.dispatch(event); |
144 } |
145 } |