import "github.com/reactivex/rxgo/v2"

func TestRxGo(t *testing.T) {
	producerCtx, cancelProducer := context.WithCancel(context.Background())
	// create a infinity producer with context for cancel
	producer := rxgo.Create(
		[]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item) {
			log.Print("create producer")
			defer func() {
				log.Print("producer done")
			}()

			i := 0
			for {
				timeout := time.After(time.Second * 1)
				select {
				case <-ctx.Done():
					return
				case <-timeout:
					next <- rxgo.Of(i)
					log.Printf("producer fire %d", i)
				}
				i = i + 1
			}
		}},
		rxgo.WithPublishStrategy(),
		rxgo.WithContext(producerCtx),
		rxgo.WithBackPressureStrategy(rxgo.Drop),
	)

	// connect to the producer
	producer.Connect(producerCtx)

	// create a child observable to receive items from producer
	append := func(index int, parent context.Context) rxgo.Observable {
		observable := rxgo.FromChannel(
			producer.Observe(),
			rxgo.WithBackPressureStrategy(rxgo.Drop),
			rxgo.WithContext(parent),
		)
		observable.DoOnNext(func(i interface{}) {
			log.Printf("Observable %d: %v", index, i)
		})

		return observable
	}

	log.Printf("sleep 5")
	time.Sleep(time.Second * 5)

	// create the first observable with own context for cancel
	observable1Ctx, cancelObservable1 := context.WithCancel(context.Background())
	append(1, observable1Ctx)

	// create the second observable with own context for cancel
	observable2Ctx, cancelObservable2 := context.WithCancel(context.Background())
	append(2, observable2Ctx)

	go func() {
		log.Printf("sleep 5")
		time.Sleep(time.Second * 5)

		log.Print("cancel 1 observable")
		cancelObservable1()

		log.Printf("sleep 5")
		time.Sleep(time.Second * 5)

		log.Print("cancel 2 observable")
		cancelObservable2()
	}()

	log.Print("wait for done")
	<-observable2Ctx.Done()
	log.Print("all observable canceled")

	log.Printf("sleep 5")
	time.Sleep(time.Second * 5)

	log.Printf("cancel the producer")
	cancelProducer()

	log.Printf("sleep 5")
	time.Sleep(time.Second * 5)
}