func TestStreamFramer_Flush(t *testing.T) {
	// Create the stream framer
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
	sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 100)
	sf.Run()

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

	f := "foo"
	fe := "bar"
	d := []byte{0xa}
	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, o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}

	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		t.Fatalf("failed to flush")
	}

	// 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 even if the frame size has not been hit, a flush will
// periodically occur.
This test checks, that even if the frame size has not been hit, a flush will
periodically occur.
// This test checks that even if the frame size has not been hit a flush will // periodically occur .
func TestStreamFramer_Flush(t *testing.T) {
	r, w := io.Pipe()
	wrappedW := &WriteCloseChecker{WriteCloser: w}
	hRate, bWindow := 100*time.Millisecond, 100*time.Millisecond
	sf := NewStreamFramer(wrappedW, false, hRate, bWindow, 100)
	sf.Run()
	dec := codec.NewDecoder(r, structs.JsonHandle)
	f := "foo"
	fe := "bar"
	d := []byte{0xa}
	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, o); err != nil {
		t.Fatalf("Send() failed %v", err)
	}
	select {
	case <-resultCh:
	case <-time.After(10 * time.Duration(testutil.TestMultiplier()) * bWindow):
		t.Fatalf("failed to flush")
	}
	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_Flush ( t * testing . T ) { r , w := io . Pipe ( ) wrappedW := & WriteCloseChecker { WriteCloser : w } hRate , bWindow := 100 * time . Millisecond , 100 * time . Millisecond sf := NewStreamFramer ( wrappedW , false , hRate , bWindow , 100 ) sf . Run ( ) dec := codec . NewDecoder ( r , structs . JsonHandle ) f := " " fe := " " d := [ ] byte { 0xa } 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 , 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 even if the frame size has not been hit, a flush will
periodically occur.
This test checks that even if the frame size has not been hit a flush will periodically occur .
// Create the stream framer // Create a decoder // Start the reader // Write only 1 byte so we do not hit the frame size // Close the reader and wait. This should cause the runner to exit