func TestStreamFramer_Batch(t *testing.T) {
	// Create the stream framer
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	// Ensure the batch window doesn't get hit
	hRate, bWindow := 100*time.Millisecond, 500*time.Millisecond
	sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 3)
	sf.Run()

	// Create a decoder
	dec := codec.NewDecoder(r, structs.JsonHandle)

	f := "foo"
	fe := "bar"
	d := []byte{0xa, 0xb, 0xc}
	o := int64(10)

	// Start the reader
	resultCh := make(chan struct{})
	go func() {
		for {
			var frame StreamFrame
			if err := dec.Decode(&frame); err != nil {
				t.Fatalf("failed to decode")
			}

			if frame.IsHeartbeat() {
				continue
			}

			if reflect.DeepEqual(frame.Data, d) && frame.Offset == o && frame.File == f && frame.FileEvent == fe {
				resultCh <- struct{}{}
				return
			}
		}
	}()

	// Write only 1 byte so we do not hit the frame size
	if err := sf.Send(f, fe, d[:1], o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}

	// Ensure we didn't get any data
	select {
	case <-resultCh:
		t.Fatalf("Got data before frame size reached")
	case <-time.After(bWindow / 2):
	}

	// Write the rest so we hit the frame size
	if err := sf.Send(f, fe, d[1:], o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}

	// Ensure we get data
	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		t.Fatalf("Did not receive data after batch size reached")
	}

	// Close the reader and wait. This should cause the runner to exit
	if err := r.Close(); err != nil {
		t.Fatalf("failed to close reader")
	}

	select {
	case <-sf.ExitCh():
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * hRate):
		t.Fatalf("exit channel should close")
	}

	sf.Destroy()
	if !wrappedW.Closed {
		t.Fatalf("writer not closed")
	}
}
// This test checks that frames will be batched till the frame size is hit (in
// the case that is before the flush).
This test checks that frames will be batched till the frame size is hit (in
the case that is before the flush).
// This test checks that frames will be batched till the frame size is hit ( in // the case that is before the flush ) .
func TestStreamFramer_Batch(t *testing.T) {
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	hRate, bWindow := 100*time.Millisecond, 500*time.Millisecond
	sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 3)
	sf.Run()
	dec := codec.NewDecoder(r, structs.JsonHandle)
	f := "foo"
	fe := "bar"
	d := []byte{0xa, 0xb, 0xc}
	o := int64(10)
	resultCh := make(chan struct{})
	go func() {
		for {
			var frame StreamFrame
			if err := dec.Decode(&frame); err != nil {
				t.Fatalf("failed to decode")
			}
			if frame.IsHeartbeat() {
				continue
			}
			if reflect.DeepEqual(frame.Data, d) && frame.Offset == o && frame.File == f && frame.FileEvent == fe {
				resultCh <- struct{}{}
				return
			}
		}
	}()
	if err := sf.Send(f, fe, d[:1], o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}
	select {
	case <-resultCh:
		t.Fatalf("Got data before frame size reached")
	case <-time.After(bWindow / 2):
	}
	if err := sf.Send(f, fe, d[1:], o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}
	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		t.Fatalf("Did not receive data after batch size reached")
	}
	if err := r.Close(); err != nil {
		t.Fatalf("failed to close reader")
	}
	select {
	case <-sf.ExitCh():
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * hRate):
		t.Fatalf("exit channel should close")
	}
	sf.Destroy()
	if !wrappedW.Closed {
		t.Fatalf("writer not closed")
	}
}
func TestStreamFramer_Batch ( t * testing . T ) { r , w := io . Pipe ( ) wrappedW := & WriteCloseChecker { WriteCloser : w } hRate , bWindow := 100 * time . Millisecond , 500 * time . Millisecond sf := NewStreamFramer ( wrappedW , false , hRate , bWindow , 3 ) sf . Run ( ) dec := codec . NewDecoder ( r , structs . JsonHandle ) f := " " fe := " " d := [ ] byte { 0xa , 0xb , 0xc } o := int64 ( 10 ) resultCh := make ( chan struct { } ) go func ( ) { for { var frame StreamFrame if err := dec . Decode ( & frame ) ; err != nil { t . Fatalf ( " " ) } if frame . IsHeartbeat ( ) { continue } if reflect . DeepEqual ( frame . Data , d ) && frame . Offset == o && frame . File == f && frame . FileEvent == fe { resultCh <- struct { } { } return } } } ( ) if err := sf . Send ( f , fe , d [ : 1 ] , o ) ; err != nil { t . Fatalf ( " " , err ) } select { case <- resultCh : t . Fatalf ( " " ) case <- time . After ( bWindow / 2 ) : } if err := sf . Send ( f , fe , d [ 1 : ] , o ) ; err != nil { t . Fatalf ( " " , err ) } select { case <- resultCh : case <- time . After ( 10 * time . Duration ( testutil . TestMultiplier ( ) ) * bWindow ) : t . Fatalf ( " " ) } if err := r . Close ( ) ; err != nil { t . Fatalf ( " " ) } select { case <- sf . ExitCh ( ) : case <- time . After ( 10 * time . Duration ( testutil . TestMultiplier ( ) ) * hRate ) : t . Fatalf ( " " ) } sf . Destroy ( ) if ! wrappedW . Closed { t . Fatalf ( " " ) } }
This test checks that frames will be batched till the frame size is hit (in
the case that is before the flush).
This test checks that frames will be batched till the frame size is hit ( in the case that is before the flush ) .
// Create the stream framer // Ensure the batch window doesn't get hit // Create a decoder // Start the reader // Write only 1 byte so we do not hit the frame size // Ensure we didn't get any data // Write the rest so we hit the frame size // Ensure we get data // Close the reader and wait. This should cause the runner to exit