Function TestSlowExecutionProcessing

db/event_manager_test.go:168–210

Test sending many events with slow-running execution to validate they get dropped after hitting the max concurrent goroutines

(t *testing.T)

// Test sending many events with slow-running execution to validate they get dropped after hitting
// the max concurrent goroutines
func TestSlowExecutionProcessing(t *testing.T) {
	ctx := base.TestCtx(t)
	terminator := make(chan bool)
	defer close(terminator)
	base.SetUpTestLogging(t, base.LevelInfo, base.KeyEvents)

	em := NewEventManager(terminator)
	em.Start(ctx, 0, -1)

	ids := make([]string, 20)
	for i := 0; i < 20; i++ {
		ids[i] = fmt.Sprintf("%d", i)
	}

	eventForTest := func(i int) (Body, string, base.Set) {
		testBody := Body{
			BodyId:  ids[i],
			"value": i,
		}
		var channelSet base.Set
		if i%2 == 0 {
			channelSet = base.SetFromArray([]string{"Even"})
		} else {
			channelSet = base.SetFromArray([]string{"Odd"})
		}
		return testBody, ids[i], channelSet
	}

	resultChannel := make(chan interface{}, 100)
	testHandler := &TestingHandler{HandledEvent: DocumentChange, handleDelay: 500}
	testHandler.SetChannel(resultChannel)
	em.RegisterEventHandler(ctx, testHandler, DocumentChange)

	for i := 0; i < 20; i++ {
		body, docid, channels := eventForTest(i % 10)
		bodyBytes := base.MustJSONMarshal(t, body)
		err := em.RaiseDocumentChangeEvent(ctx, bodyBytes, docid, "", channels, false)
		assert.NoError(t, err)
	}

	assertChannelLengthWithTimeout(t, resultChannel, 20, 10*time.Second)

}

func TestCustomHandler(t *testing.T) {
	ctx := base.TestCtx(t)