// SubscribeAll will create a subscription which can be used to stream all events in an
// orderly fashion. This mechanism should probably be mostly useful for building projections
SubscribeAll will create a subscription which can be used to stream all events in an
orderly fashion. This mechanism should probably be mostly useful for building projections
// SubscribeAll will create a subscription which can be used to stream all events in an // orderly fashion . This mechanism should probably be mostly useful for building projections
func (es *EventStore) SubscribeAll(ctx context.Context, opts ...SubAllOpt) (Subscription, error) {
	cfg := SubAllConfig{
		offset:       0,
		batchSize:    100,
		pollInterval: 100 * time.Millisecond,
	}
	for _, opt := range opts {
		cfg = opt(cfg)
	}
	if cfg.batchSize < 1 {
		return Subscription{}, fmt.Errorf("batch size should be at least 1")
	}
	sub := Subscription{
		Err:       make(chan error, 1),
		EventData: make(chan EventData, cfg.batchSize),
		close:     make(chan struct{}, 1),
	}
	go func() {
		var done error
		for {
			select {
			case <-sub.close:
				sub.Err <- ErrSubscriptionClosedByClient
				return
			case <-ctx.Done():
				sub.Err <- ctx.Err()
				return
			case <-time.After(cfg.pollInterval):
				if done != nil {
					if len(sub.EventData) != 0 {
						break
					}
					sub.Err <- done
					return
				}
				var evts []gormEvent
				if err := es.db.
					Where("id > ?", cfg.offset).
					Order("id asc").
					Limit(cfg.batchSize).
					Find(&evts).Error; err != nil {
					done = err
					break
				}
				if len(evts) == 0 {
					sub.Err <- io.EOF
					break
				}
				cfg.offset = cfg.offset + len(evts)
				decoded, err := es.decodeEvts(evts)
				if err != nil {
					done = err
					break
				}
				for _, evt := range decoded {
					sub.EventData <- evt
				}
			}
		}
	}()
	return sub, nil
}
func ( es * EventStore ) SubscribeAll ( ctx context . Context , opts ... SubAllOpt ) ( Subscription , error ) { cfg := SubAllConfig { offset : 0 , batchSize : 100 , pollInterval : 100 * time . Millisecond , } for _ , opt := range opts { cfg = opt ( cfg ) } if cfg . batchSize < 1 { return Subscription { } , fmt . Errorf ( " " ) } sub := Subscription { Err : make ( chan error , 1 ) , EventData : make ( chan EventData , cfg . batchSize ) , close : make ( chan struct { } , 1 ) , } go func ( ) { var done error for { select { case <- sub . close : sub . Err <- ErrSubscriptionClosedByClient return case <- ctx . Done ( ) : sub . Err <- ctx . Err ( ) return case <- time . After ( cfg . pollInterval ) : if done != nil { if len ( sub . EventData ) != 0 { break } sub . Err <- done return } var evts [ ] gormEvent if err := es . db . Where ( " " , cfg . offset ) . Order ( " " ) . Limit ( cfg . batchSize ) . Find ( & evts ) . Error ; err != nil { done = err break } if len ( evts ) == 0 { sub . Err <- io . EOF break } cfg . offset = cfg . offset + len ( evts ) decoded , err := es . decodeEvts ( evts ) if err != nil { done = err break } for _ , evt := range decoded { sub . EventData <- evt } } } } ( ) return sub , nil }
SubscribeAll will create a subscription which can be used to stream all events in an
orderly fashion.
SubscribeAll will create a subscription which can be used to stream all events in an orderly fashion .
// Make sure client reads all buffered events