ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
if (ret == EOF)
return EOF;
+ else if (ret == EAGAIN)
+ /* Stream is inactive for now (live reading). */
+ return EAGAIN;
else if (ret) {
fprintf(stderr, "[error] Reading event failed.\n");
return ret;
int i, ret;
stream_pos = &cfs->pos;
- for (i = 0; i < stream_pos->packet_real_index->len; i++) {
- index = &g_array_index(stream_pos->packet_real_index,
+ for (i = 0; i < stream_pos->packet_index->len; i++) {
+ index = &g_array_index(stream_pos->packet_index,
struct packet_index, i);
- if (index->timestamp_end < timestamp)
+ if (index->ts_real.timestamp_end < timestamp)
continue;
stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
* either find at least one event, or we reach the first packet
* (some packets can be empty).
*/
- for (i = stream_pos->packet_real_index->len - 1; i >= 0; i--) {
+ for (i = stream_pos->packet_index->len - 1; i >= 0; i--) {
stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
count = 0;
/* read each event until we reach the end of the stream */
if (ret == EOF) {
ret = 0;
continue;
- } else if (ret) {
+ } else if (ret != 0 && ret != EAGAIN) {
goto error;
}
/* Add to heap */
assert(removed == file_stream);
ret = 0;
goto end;
+ } else if (ret == EAGAIN) {
+ /*
+ * Live streaming: the stream is inactive for now, we
+ * just updated the timestamp_end to skip over this
+ * stream up to a certain point in time.
+ *
+ * Since we can't guarantee that a stream will ever have
+ * any activity, we can't rely on the fact that
+ * bt_iter_next will be called for each stream and deal
+ * with inactive streams. So instead, we return 0 here
+ * to the caller and let the read API handle the
+ * retry case.
+ */
+ ret = 0;
+ goto reinsert;
} else if (ret) {
goto end;
}
+
+reinsert:
/* Reinsert the file stream into the heap, and rebalance. */
removed = bt_heap_replace_max(iter->stream_heap, file_stream);
assert(removed == file_stream);
-
end:
return ret;
}