Function joinTest

observable_operator_test.go:990–1016

(ctx context.Context, t *testing.T, left, right []interface{}, window Duration, expected []int64)

}

func joinTest(ctx context.Context, t *testing.T, left, right []interface{}, window Duration, expected []int64) {
	leftObs := testObservable(ctx, left...)
	rightObs := testObservable(ctx, right...)

	obs := leftObs.Join(func(ctx context.Context, l, r interface{}) (interface{}, error) {
		return map[string]interface{}{
			"l": l,
			"r": r,
		}, nil
	},
		rightObs,
		func(i interface{}) time.Time {
			return time.Unix(0, i.(map[string]int64)["tt"]*1000000)
		},
		window,
	)

	Assert(ctx, t, obs, CustomPredicate(func(items []interface{}) error {
		actuals := make([]int64, 0)
		for _, p := range items {
			val := p.(map[string]interface{})
			actuals = append(actuals, val["l"].(map[string]int64)["V"], val["r"].(map[string]int64)["V"])
		}
		assert.Equal(t, expected, actuals)
		return nil
	}))
}

func Test_Observable_Join1(t *testing.T) {
	defer goleak.VerifyNone(t)