Method SubscribeWithHistory

pkg/events/redis/store.go:346–447

SubscribeWithHistory is like FetchHistory, but after fetching historical events, this continues sending live events until the context is done.

(
	ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int, hdl events.Handler,
)

// SubscribeWithHistory is like FetchHistory, but after fetching historical events,
// this continues sending live events until the context is done.
func (ps *PubSubStore) SubscribeWithHistory(
	ctx context.Context, names []string, ids []*ttnpb.EntityIdentifiers, after *time.Time, tail int, hdl events.Handler,
) (err error) {
	start := "0-0"
	switch {
	case after == nil:
	case after.IsZero():
		after = nil
	default:
		// Truncate to milliseconds to be consistent with the JSON API.
		afterMS := after.Truncate(time.Millisecond)
		after = &afterMS
		// Account for a clock skew on the Redis server of up to 1 second.
		start = formatStreamTime(after.Add(-1 * time.Second))
	}

	states := make([]*streamState, len(ids))
	for i, id := range ids {
		states[i] = &streamState{
			id:     id,
			stream: ps.eventStream(ctx, id),
			start:  start,
		}
	}

	for _, s := range states {
		evtPBs, nextStart, err := ps.tailStream(ctx, names, s.id, s.start, tail)
		if err != nil {
			return err
		}
		for _, evtPB := range evtPBs {
			if after != nil {
				if evtTime := ttnpb.StdTime(evtPB.GetTime()); evtTime != nil &&
					!evtTime.Truncate(time.Millisecond).After(*after) {
					continue
				}
			}
			evt, err := events.FromProto(evtPB)
			if err != nil {
				return err
			}
			hdl.Notify(evt)
		}
		s.start = nextStart
	}

	matchNames := xMessageHasEventName(names...)

	// We expect that the subscriber can ingest `tail` events at a time.
	eventCountLimit := int64(tail)
	switch {
	case eventCountLimit < 8:
		eventCountLimit = 8
	case eventCountLimit > 1024:
		eventCountLimit = 1024
	}

	ch := make(chan []redis.XStream, 1)