// doTestDispatchInBatches tests dispatch() method using varying |batchSize|s.
doTestDispatchInBatches tests dispatch() method using varying |batchSize|s.
// doTestDispatchInBatches tests dispatch () method using varying |batchSize|s .
func doTestDispatchInBatches(t *testing.T, useMemStore bool) {
	const num = 40
	const currentDayIndex = 10
	for _, batchSize := range []int{1, 10, 20, 40, 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("BatchSize: [%d], got observations [%v], expected [%v]", batchSize, len(obVals), num)
		}
		d := newTestDispatcher(store, batchSize, 0)
		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 >= d.batchSize {
			if len(analyzer.obBatch) != num/d.batchSize {
				t.Errorf("BatchSize: [%d], unexpected number of observations dispatched, got [%v] expected [%v]", d.batchSize, len(analyzer.obBatch), num)
			}
		} else if len(analyzer.obBatch) != 1 {
			t.Errorf("BatchSize: [%d], expected all observations to be sent at once, got [%v] expected [1]", d.batchSize, len(analyzer.obBatch))
		}
		if num >= d.batchSize {
			if analyzer.numSent != num/d.batchSize {
				t.Errorf("BatchSize: [%d], unexpected number of analyzer send calls, got [%d], want [%d]", d.batchSize, analyzer.numSent, num/d.batchSize)
			}
		} else if analyzer.numSent != 1 {
			t.Errorf("BatchSize: [%d], expected all observations to be sent at once, got [%d], want [1]", d.batchSize, analyzer.numSent)
		}
		if obValsLen, _ := d.store.GetNumObservations(key); obValsLen != 0 {
			t.Errorf("BatchSize: [%d], got [%d] observations, expected [0] observations in the store for meatdata [%v]", d.batchSize, obValsLen, key)
		}
		analyzer.numSent = 0
		analyzer.obBatch = nil
		storage.ResetStoreForTesting(d.store, true)
	}
}
func doTestDispatchInBatches ( t * testing . T , useMemStore bool ) { const num = 40 const currentDayIndex = 10 for _ , batchSize := range [ ] int { 1 , 10 , 20 , 40 , 100 } { store , key , obVals , err := makeTestStore ( num , currentDayIndex , useMemStore ) if err != nil { t . Fatalf ( " " , err ) } if len ( obVals ) != num { t . Errorf ( " " , batchSize , len ( obVals ) , num ) } d := newTestDispatcher ( store , batchSize , 0 ) 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 >= d . batchSize { if len ( analyzer . obBatch ) != num / d . batchSize { t . Errorf ( " " , d . batchSize , len ( analyzer . obBatch ) , num ) } } else if len ( analyzer . obBatch ) != 1 { t . Errorf ( " " , d . batchSize , len ( analyzer . obBatch ) ) } if num >= d . batchSize { if analyzer . numSent != num / d . batchSize { t . Errorf ( " " , d . batchSize , analyzer . numSent , num / d . batchSize ) } } else if analyzer . numSent != 1 { t . Errorf ( " " , d . batchSize , analyzer . numSent ) } if obValsLen , _ := d . store . GetNumObservations ( key ) ; obValsLen != 0 { t . Errorf ( " " , d . batchSize , obValsLen , key ) } analyzer . numSent = 0 analyzer . obBatch = nil storage . ResetStoreForTesting ( d . store , true ) } }
doTestDispatchInBatches tests dispatch() method using varying |batchSize|s.
doTestDispatchInBatches tests dispatch () method using varying |batchSize|s .
// Test dispatch for different batch sizes // Recreate the test store with |num| entries for every run as they get // deleted after every successful dispatch attempt. // run dispatcher by setting threshold and frequency to "0" and different // batchsizes. // Assert that last timestamp has been modified to the current time. // check if all msgs are sent // check if the data is sent in batches based on batchSize // check if all the sent msgs are deleted from the Shuffler datastore // reset analyzer // reset store