Method tailStream

pkg/events/redis/store.go:176–206

(
	ctx context.Context, names []string, ids *ttnpb.EntityIdentifiers, start string, tail int,
)

}

func (ps *PubSubStore) tailStream(
	ctx context.Context, names []string, ids *ttnpb.EntityIdentifiers, start string, tail int,
) (eventPBs []*ttnpb.Event, nextStart string, err error) {
	matchNames := xMessageHasEventName(names...)
	if tail > 0 {
		msgs, err := ps.client.XRevRangeN(ctx, ps.eventStream(ctx, ids), "+", start, int64(tail)).Result()
		if err != nil {
			return nil, "", ttnredis.ConvertError(err)
		}
		if len(msgs) == 0 {
			return nil, start, nil
		}
		eventPBs = eventsFromXMessages(msgs, matchNames)
		reverseEvents(eventPBs)
		nextStart = msgs[0].ID
	} else {
		msgs, err := ps.client.XRange(ctx, ps.eventStream(ctx, ids), start, "+").Result()
		if err != nil {
			return nil, "", ttnredis.ConvertError(err)
		}
		if len(msgs) == 0 {
			return nil, start, nil
		}
		eventPBs = eventsFromXMessages(msgs, matchNames)
		nextStart = msgs[len(msgs)-1].ID
	}
	if err = ps.loadEventData(ctx, ps.client, eventPBs...); err != nil {
		return nil, "", err
	}
	return eventPBs, nextStart, nil
}

// FetchHistory fetches the tail (optional) of historical events matching the given
// names (optional) and identifiers (mandatory) after the given time (optional).