// doTestDispatchInBatches tests dispatch() method using varying |threshold|
// limits.
doTestDispatchInBatches tests dispatch() method using varying |threshold|
limits.
// doTestDispatchInBatches tests dispatch () method using varying |threshold| // limits .
func doTestDispatchBasedOnThresholds(t *testing.T, useMemStore bool) {
	const num = 40
	const currentDayIndex = 10
	for _, threshold := range []int{1, 10, 20, 40, 80, 100} {
		store, key, obVals, err := makeTestStore(num, currentDayIndex, useMemStore)
		if err != nil {
			t.Fatalf("got error [%v] in test store setup", err)
		}
		if len(obVals) != num {
			t.Errorf("Threshold: [%d], got observations [%v], expected [%v]", threshold, len(obVals), num)
		}
		d := newTestDispatcher(store, num, threshold)
		analyzer := getAnalyzerTransport(d)
		d.dispatch(1 * time.Millisecond)
		now := time.Now()
		if uint32(d.lastDispatchTime.Sub(now).Minutes()) != 0 {
			t.Errorf("got last dispatch time [%v], expected last dispatch time [%v] to be updated to current", d.lastDispatchTime, now)
		}
		if num < threshold {
			if analyzer.numSent != 0 {
				t.Errorf("Threshold: [%d], unexpected number of analyzer send calls, got [%d], want [0]", threshold, analyzer.numSent)
			}
		} else {
			if len(analyzer.obBatch) != 1 {
				t.Errorf("Threshold: [%d], unexpected number of observations dispatched, got [%v] expected [1]", threshold, len(analyzer.obBatch))
			}
			if obValsLen, _ := store.GetNumObservations(key); obValsLen != 0 {
				t.Errorf("Threshold: [%d], got [%d] observations, expected [0] observations in the store for meatdata [%v]", threshold, obValsLen, key)
			}
		}
		analyzer.numSent = 0
		analyzer.obBatch = nil
		storage.ResetStoreForTesting(store, true)
	}
}
func doTestDispatchBasedOnThresholds ( t * testing . T , useMemStore bool ) { const num = 40 const currentDayIndex = 10 for _ , threshold := range [ ] int { 1 , 10 , 20 , 40 , 80 , 100 } { store , key , obVals , err := makeTestStore ( num , currentDayIndex , useMemStore ) if err != nil { t . Fatalf ( " " , err ) } if len ( obVals ) != num { t . Errorf ( " " , threshold , len ( obVals ) , num ) } d := newTestDispatcher ( store , num , threshold ) analyzer := getAnalyzerTransport ( d ) d . dispatch ( 1 * time . Millisecond ) now := time . Now ( ) if uint32 ( d . lastDispatchTime . Sub ( now ) . Minutes ( ) ) != 0 { t . Errorf ( " " , d . lastDispatchTime , now ) } if num < threshold { if analyzer . numSent != 0 { t . Errorf ( " " , threshold , analyzer . numSent ) } } else { if len ( analyzer . obBatch ) != 1 { t . Errorf ( " " , threshold , len ( analyzer . obBatch ) ) } if obValsLen , _ := store . GetNumObservations ( key ) ; obValsLen != 0 { t . Errorf ( " " , threshold , obValsLen , key ) } } analyzer . numSent = 0 analyzer . obBatch = nil storage . ResetStoreForTesting ( store , true ) } }
doTestDispatchInBatches tests dispatch() method using varying |threshold|
limits.
doTestDispatchInBatches tests dispatch () method using varying |threshold| limits .
// Test dispatch for different thresholds // Recreate the test store with |num| entries for every run as they get // deleted after every successful dispatch attempt. // run dispatcher with frequency set to "0" and batchsize set to the max // chunk size - "num" for sending all messages at once in one large batch. // Assert that last timestamp has been modified to the current time. // check if all msgs are sent // make sure that all sent msgs are deleted from the Shuffler datastore // reset analyzer // reset store