func init() {
	register.Function2x0(produceFn)
	register.Function4x0(getPanes)
	register.Emitter1[int]()
}

func produceFn(_ []byte, emit func(beam.EventTime, int)) {
	baseT := mtime.Now()
	for i := 0; i < 10; i++ {
		emit(baseT.Add(time.Minute), i)
	}
}

func Produce(s beam.Scope) beam.PCollection {
	return beam.ParDo(s, produceFn, beam.Impulse(s))
}

func getPanes(ctx context.Context, pi typex.PaneInfo, _ int, emit func(int)) {
	log.Output(ctx, log.SevWarn, 0, fmt.Sprintf("got pane %+v", pi))
	emit(int(pi.Index))
}
func TestPanes(t *testing.T) {
	p, scp := beam.NewPipelineWithRoot()

	c := Produce(scp)
	windowed := beam.WindowInto(
		scp,
		window.NewFixedWindows(5*time.Minute),
		c,
		beam.Trigger(trigger.AfterEndOfWindow().
			EarlyFiring(
				trigger.Repeat(
					trigger.AfterCount(2),
				),
			),
		),
		beam.PanesDiscard(),
	)
	panes := beam.ParDo(scp, getPanes, windowed)
	paneIdxs := beam.WindowInto(scp, window.NewGlobalWindows(), panes)
	passert.Count(scp, paneIdxs, "pane idxs", 10)
	passert.EqualsList(scp, paneIdxs, []int{0, 0, 1, 1, 2, 0, 0, 1, 1, 2})
	ptest.RunAndValidate(t, p)
}